Vrismcost 0.2.6 · every line
Nine files, 4,128 lines. Each one sits at its own address and is served as plain text, so you can open it, link to it, curl it or diff it without downloading anything. Every checksum below is the file being served.
Take all nine and check them, in two commands
curl --remote-name-all "https://your-site/src/{vrismcost.py,fetcher.py,engine.py,price.py,report.py,vault.py,test_vrismcost.py,README.md,requirements.txt,SHA256SUMS}"
sha256sum -c SHA256SUMS
The manifest is in the format sha256sum -c already understands. On macOS, shasum -a 256 -c SHA256SUMS. On Windows, certutil -hashfile <file> SHA256 checks one at a time.
"""
Vrismcost — cost basis and realised P&L for Robinhood Chain.
Reads your wallets straight off the public RPC and computes a cost basis on
this machine. No account, no API key, no server: the only thing that ever
leaves your computer is the address you asked about, and it goes to the node
and nowhere else.
python vrismcost.py 0xabc... one wallet, full history
python vrismcost.py 0xabc... 0xdef... a wallet set, netted together
python vrismcost.py 0xabc... --recent=2000000 bound the scan
python vrismcost.py 0xabc... --quote=USDG name the quote asset
python vrismcost.py 0xabc... --eth=3455.20 fix the ETH rate yourself
python vrismcost.py 0xabc... --eth=off leave ETH legs unpriced
python vrismcost.py 0xabc... --json=out.json write the full report
python vrismcost.py 0xabc... --csv=2026 write the schedule for an accountant
python vrismcost.py 0xabc... --method=HIFO choose the matching method
python vrismcost.py 0xabc... --rpc=http://… read from your own node
python vrismcost.py --vault scan the wallets in your vault
python vrismcost.py vault init create the encrypted vault
python vrismcost.py vault status what it holds
python vrismcost.py vault add 0xabc... main register a wallet
python vrismcost.py vault answer <tx> own 2568.72 2026-07-01 answer the queue
python vrismcost.py vault answer <tx> income 84.60
python vrismcost.py vault answer <tx> exclude
python vrismcost.py vault forget <tx> undo one answer
python vrismcost.py vault recover use the recovery code
What it will not do: guess a price, book a wallet move as a sale, read a split
as a disposal, or quietly drop a transaction it could not price. Anything it
cannot prove lands in the queue with the reason.
"""
from __future__ import annotations
import json
import sys
import time
from decimal import Decimal as D
from fetcher import Fetcher, LAUNCH_BLOCK, token_metadata
from engine import Engine
from price import PriceOracle
from report import write_report
# ---------------------------------------------------------------- pipeline
QUOTE_CANDIDATES = ("USDG", "USDC", "USDT", "DAI")
def run(addresses: list[str], *, start: int | None = None, end: int | None = None,
recent: int | None = None, quote: str | None = None,
eth_mode: str = "derive", native_price=None, answers=None,
fetcher: Fetcher | None = None, verbose: bool = False, stage=None,
rpc: str | None = None) -> dict:
"""The whole pipeline, with nothing printed: read the chain, name the
tokens, pick the quote asset, price ETH, match the lots.
Kept apart from main() so the pipeline can be exercised on its own: the
tests run it end to end against a node held in memory, and main() adds
only the vault, the flags and the printing.
stage(name, **info) is told when each phase begins: "scan" with start and
head, then "tokens", "price" (only when the rate is being derived), and
"match". report is None when the range held no Transfer activity.
"""
f = fetcher or (Fetcher(rpc=rpc, verbose=verbose) if rpc else Fetcher(verbose=verbose))
say = stage or (lambda name, **info: None)
head = f.head() if end is None else end
if start is None:
start = max(LAUNCH_BLOCK, head - recent) if recent else LAUNCH_BLOCK
say("scan", start=start, head=head)
scan = f.scan(addresses, start=start, end=head)
txs = scan["transactions"]
result = {"start": start, "head": head, "scan": scan, "quote": quote,
"tokens": {}, "report": None, "oracle": None, "ethMode": eth_mode}
if not txs:
return result
say("tokens")
tokens = sorted({l.token for t in txs for l in t.legs})
meta = token_metadata(f, tokens)
# The quote asset is whichever stablecoin this wallet actually trades
# against. Guessing it wrong silently reclassifies every buy, so it is
# detected from the tokens present and can always be overridden.
if not quote:
symbols = {m["symbol"].upper() for m in meta.values()}
quote = next((c for c in QUOTE_CANDIDATES if c in symbols), "USDG")
# Most real trades on this chain settle in ETH, so without a rate the
# report covers almost nothing. The rate is read back off the chain from
# WETH/USDG swaps rather than a feed, which keeps the wallet private and
# works for blocks whose state was pruned long ago.
oracle = None
if eth_mode == "derive":
say("price")
oracle = PriceOracle(f, bucket=100_000)
native_price = oracle
say("match")
engine = Engine(addresses, meta, quote_symbol=quote,
native_price=native_price, answers=answers)
result.update(quote=quote, tokens=meta, oracle=oracle, report=engine.run(txs))
return result
def report_meta(result: dict) -> dict:
"""The header block of an export: which chain, which blocks, where the ETH
rate came from."""
oracle = result["oracle"]
ometa = oracle.summary() if oracle is not None else None
return {
"chainName": "Robinhood Chain", "chainId": 4663,
"fromBlock": result["start"], "toBlock": result["head"],
"ethRateSource": ("derived from WETH/USDG pools on chain"
if oracle is not None else
("supplied by you" if result["ethMode"] == "fixed" else "none")),
"ethRateWindows": (f"{ometa['resolved']}/{ometa['buckets']} resolved, "
f"{ometa['bucketBlocks']:,} blocks each") if ometa else "",
}
def human(n: str | D, dp: int = 2) -> str:
return f"{D(str(n)):,.{dp}f}"
def bar(fraction: float, width: int = 28) -> str:
filled = int(round(fraction * width))
return "#" * filled + "." * (width - filled)
# ---------------------------------------------------------------- vault CLI
def _ask(prompt: str) -> str:
import getpass
return getpass.getpass(prompt)
def vault_command(argv: list[str]) -> int:
"""Vault subcommands. The passphrase is read from the terminal and never
written anywhere, and the vault is locked again the moment the task ends."""
# Imported here, not at the top: scanning never touches the vault, and a
# scan should run where the vault's own dependencies are not installed.
from vault import Vault, VaultError, WrongPassphrase
sub = argv[2] if len(argv) > 2 else "status"
if sub == "init":
pw = _ask("Choose a passphrase (10+ characters): ")
if pw != _ask("Repeat it: "):
print("Those did not match.")
return 1
try:
v, code = Vault.create(pw)
except VaultError as e:
print(e)
return 1
print(f"\nVault created at {v.path}\n")
print("RECOVERY CODE — write this down, on paper, now:\n")
print(f" {code}\n")
print("It is the only way back in if you forget the passphrase. It is not")
print("stored anywhere, and it will not be shown again.")
v.lock()
return 0
if sub == "recover":
code = input("Recovery code: ").strip()
pw = _ask("New passphrase: ")
if pw != _ask("Repeat it: "):
print("Those did not match.")
return 1
try:
v, new_code = Vault.recover(code, pw)
except VaultError as e:
print(e)
return 1
print("\nVault recovered. Your old recovery code no longer works.\n")
print(f"NEW RECOVERY CODE:\n\n {new_code}\n")
v.lock()
return 0
try:
v = Vault.unlock(_ask("Passphrase: "))
except (VaultError, WrongPassphrase) as e:
print(e)
return 1
try:
if sub == "status":
s = v.summary()
print(f"vault {s['path']}")
print(f"id {s['vaultId']} revision {s['revision']}")
print(f"size {s['bytes']} bytes on disk, fully encrypted")
print(f"wallets {s['wallets']}")
print(f"answers {s['answers']}")
print(f"rates {s['rates']}")
if v.data["wallets"]:
print("\nwallets:")
for w in v.data["wallets"]:
print(f" {w['address']} {w['label']}")
if v.answers():
print("\nanswers:")
for tx, a in list(v.answers().items())[:12]:
val = f" = {a['value']}" if a.get("value") else ""
print(f" {tx[:20]}… {a['action']}{val}")
return 0
if sub == "add":
addr = argv[3]
label = argv[4] if len(argv) > 4 else ""
v.add_wallet(addr, label)
v.save()
print(f"registered {addr.lower()} {('as ' + label) if label else ''}")
return 0
if sub == "answer":
tx, action = argv[3], argv[4]
value = argv[5] if len(argv) > 5 else None
date = argv[6] if len(argv) > 6 else None
if action not in Engine.ANSWER_KINDS:
print(f"Unknown action. Try one of: {', '.join(Engine.ANSWER_KINDS)}")
return 1
if date is not None:
try:
time.strptime(date, "%Y-%m-%d")
except ValueError:
print("The date must look like 2026-07-01.")
return 1
v.answer(tx, action, value, date=date)
v.save()
print(f"answered {tx[:20]}… as {action}" + (f" = {value}" if value else "")
+ (f", acquired {date}" if date else ""))
return 0
if sub == "forget":
v.unanswer(argv[3])
v.save()
print(f"forgot {argv[3][:20]}…")
return 0
print(f"Unknown vault command: {sub}")
return 1
finally:
v.lock()
def main(argv: list[str]) -> int:
if len(argv) > 1 and argv[1] == "vault":
return vault_command(argv)
addresses = [a.lower() for a in argv[1:]
if a.startswith("0x") and len(a) == 42]
use_vault = "--vault" in argv[1:]
# The vault supplies the wallet set and every answer given to the queue, so
# a correction made once is a correction made for good.
vault = answers = None
if use_vault:
from vault import Vault, VaultError, WrongPassphrase
try:
vault = Vault.unlock(_ask("Passphrase: "))
except (VaultError, WrongPassphrase) as e:
print(e)
return 1
addresses = addresses or vault.wallets()
answers = vault.answers()
if not addresses:
print(__doc__.strip())
return 1
recent = quote = out_path = csv_prefix = rpc = None
method = "FIFO"
native_price = None
eth_mode = "derive" # read the rate off the chain unless told otherwise
for a in argv[1:]:
if a.startswith("--recent="):
recent = int(a.split("=", 1)[1])
elif a.startswith("--quote="):
quote = a.split("=", 1)[1].upper()
elif a.startswith("--json="):
out_path = a.split("=", 1)[1]
elif a.startswith("--csv="):
csv_prefix = a.split("=", 1)[1]
elif a.startswith("--rpc="):
rpc = a.split("=", 1)[1]
elif a.startswith("--method="):
method = a.split("=", 1)[1].upper()
elif a.startswith("--eth="):
v = a.split("=", 1)[1].lower()
native_price = None if v in ("off", "none", "no") else D(v)
eth_mode = "off" if native_price is None else "fixed"
if vault is not None and eth_mode == "derive":
stored = vault.setting("ethRate")
if stored:
native_price, eth_mode = D(str(stored)), "fixed"
print(f"Vrismcost · Robinhood Chain · chain 4663", file=sys.stderr)
def stage(name: str, **info) -> None:
if name == "scan":
print(f"scanning {len(addresses)} wallet(s), "
f"blocks {info['start']:,}-{info['head']:,}\n", file=sys.stderr)
elif name == "price":
print("deriving the ETH rate from WETH/USDG pools \u2026", file=sys.stderr)
began = time.time()
result = run(addresses, recent=recent, quote=quote, eth_mode=eth_mode,
native_price=native_price, answers=answers, verbose=True, stage=stage,
rpc=rpc)
if result["report"] is None:
print("No Transfer activity found for these addresses in that range.")
return 0
report, meta, quote = result["report"], result["tokens"], result["quote"]
scan, oracle = result["scan"], result["oracle"]
tokens = sorted(meta)
elapsed = time.time() - began
# ---------------------------------------------------------------- output
cov = report["coverage"]
print("=" * 66)
print(f" {len(addresses)} wallet(s) · {report['transactions']} transactions "
f"· {len(tokens)} tokens · quote {quote}")
print("=" * 66)
print("\nTOKENS HELD OR TRADED")
for t in tokens:
m = meta[t]
mark = " " if m["resolved"] else "?"
print(f" {mark} {m['symbol']:<10} {m['decimals']:>2}d {m['name'][:38]:<38} {t[:12]}…")
print("\nWHAT THE TRANSACTIONS WERE")
for kind, n in sorted(report["shapes"].items(), key=lambda kv: -kv[1]):
print(f" {kind:<13} {n:>4}")
answered = cov.get("answered", 0)
print(f"\nCOVERAGE {bar((cov['full'] + answered) / max(1, cov['total']))} "
f"{cov['pct']}%")
print(f" {cov['full']} proven by the chain \u00b7 {answered} answered by you "
f"\u00b7 {cov['blocked']} still open")
if cov["blocked"]:
print(f" {cov['blocked']} transaction(s) need you \u2014 an export now is marked INCOMPLETE")
if vault is None:
print(" (run with --vault to answer them once and keep the answers)")
if report["lots"]:
print(f"\nLOTS BUILT ({len(report['lots'])})")
for l in report["lots"][:12]:
tag = " [split]" if l["split"] else ""
print(f" #{l['n']:<3} {human(l['qty'], 4):>20} {l['symbol']:<8} "
f"@ {l['unitCost']:>16} cost {human(l['cost']):>12} "
f"{l['acquired']}{tag}")
if len(report["lots"]) > 12:
print(f" … {len(report['lots']) - 12} more")
else:
print("\nLOTS BUILT (0)")
print(" No acquisition in this history states its own price, so no lot")
print(" could be opened from chain evidence alone. Everything is queued.")
for d in report["disposals"]:
proceeds = f"{human(d['proceeds'])} {quote}" if d["proceeds"] is not None else "UNPRICED"
print(f"\nDISPOSAL {human(d['qty'], 6)} {d['symbol']} on {d['at']} "
f"proceeds {proceeds} ({d['source']})")
for name in ("FIFO", "LIFO", "HIFO", "AVERAGE"):
m = d["methods"][name]
extra = f" unmatched {m['unmatched']}" if D(m.get("unmatched", "0")) > 0 else ""
gain = human(m["gain"]) if m["gain"] is not None else "UNPRICED"
print(f" {name:<8} basis {human(m['costBasis']):>14} "
f"gain {gain:>14} legs {len(m['legs']):>2}{extra}")
if report["corporateActions"]:
print("\nCORPORATE ACTIONS")
for c in report["corporateActions"]:
print(f" {c['at']} {c['type']} {c['ratio']}:1 on {c['token']} "
f"→ lots {c['lots']} adjusted, basis and dates preserved")
if report["queue"]:
print(f"\nCORRECTION QUEUE ({len(report['queue'])})")
for q in report["queue"][:10]:
given = ", ".join(f"{v} {k}" for k, v in q["given"].items()) or "\u2014"
gained = ", ".join(f"{v} {k}" for k, v in q["gained"].items()) or "\u2014"
mark = "\u2713" if q.get("answer") else " "
print(f" {mark} {q['at']} {q['kind']:<12} {given} \u2192 {gained}")
if q.get("answer"):
a = q["answer"]
val = f" = {a['value']}" if a.get("value") else ""
print(f" answered: {a['label']}{val}")
else:
print(f" {q['note'][:74]}")
print(f" vrismcost.py vault answer {q['tx'][:18]}… <action> [value]")
if len(report["queue"]) > 10:
print(f" … {len(report['queue']) - 10} more")
if oracle is not None:
s = oracle.summary()
print(f"\nETH RATE derived from the chain \u00b7 {s['resolved']}/{s['buckets']} "
f"windows resolved \u00b7 {s['bucketBlocks']:,} blocks each")
for r in s["rates"][:6]:
print(f" block {r['block']:>12,} {human(r['rate']):>10} "
f"{r['samples']:>4} swaps spread {r['spreadPct']}% {r['confidence']}")
if s["unresolved"]:
print(f" {s['unresolved']} window(s) unresolved "
f"({'; '.join(s['unresolvedReasons'])}) \u2014 those ETH legs stay unpriced")
elif eth_mode == "fixed":
print(f"\nETH RATE {human(native_price)} {quote} \u00b7 supplied by you")
else:
print(f"\nETH RATE none \u2014 ETH legs left unpriced")
print(f"\nREALISED (FIFO) {human(report['realised'])} {quote}")
if report.get("income"):
valued = [i for i in report["income"] if i["value"] is not None]
total = sum((D(i["value"]) for i in valued), D("0"))
print(f"INCOME {human(total)} {quote} \u00b7 "
f"{len(report['income'])} confirmed by you"
+ (f", {len(report['income']) - len(valued)} without a value"
if len(valued) < len(report["income"]) else ""))
print(f"OPEN LOTS {report['openLots']}")
if report.get("unmatched"):
print(f"UNMATCHED {len(report['unmatched'])} disposal(s) sold more than the scanned "
f"history bought \u2014 widen the scan or answer them")
if report.get("excluded"):
print(f"EXCLUDED {len(report['excluded'])} transaction(s), at your instruction")
if report.get("gasPaidByOthers"):
print(f"GAS NOT YOURS {report['gasPaidByOthers']} transaction(s) sent by another "
f"address \u2014 their gas is left out of your figures")
print(f"\nscanned in {elapsed:.1f}s · {scan['rpcCalls']:,} rpc calls · "
f"{scan['roundTrips']:,} round trips · nothing left this machine")
if vault is not None:
vault.lock()
if csv_prefix:
written = write_report(csv_prefix, report, report_meta(result), method)
blocked = report["coverage"].get("blocked", 0)
print(f"\nEXPORT {method} \u00b7 "
+ ("complete" if blocked == 0
else f"INCOMPLETE, {blocked} transaction(s) unresolved"))
for path, n in written:
print(f" {path.name:<34} {n:>4} row(s)")
if blocked:
print(" Unresolved transactions are in the schedule marked UNPRICED,")
print(" not dropped. Answer them with `vault answer` and export again.")
if out_path:
payload = {k: v for k, v in report.items() if k != "events"}
payload["tokens"] = meta
payload["scan"] = {k: v for k, v in scan.items() if k != "transactions"}
with open(out_path, "w", encoding="utf-8") as fh:
json.dump(payload, fh, indent=2, default=str)
print(f"report written to {out_path}")
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv))
"""
Vrismcost — chain fetcher.
Turns a list of addresses into the exact input the cost-basis engine wants:
every Transfer leg that touches those addresses, grouped by transaction, with
the block timestamp, the native ETH the transaction moved, and the gas it paid.
Standard library only. No API key, no account, no third-party service — the
public RPC answers everything, so nothing about a user's wallets is ever sent
anywhere except the node itself.
Four measured facts about Robinhood Chain shape every decision here:
1. The RPC sits behind Cloudflare and rejects default library user agents with
403. A browser-shaped User-Agent header is mandatory, not cosmetic.
2. eth_getLogs refuses any query matching more than 10,000 logs — as an
explicit error, never a silent truncation. Ranges are bisected on it.
3. Topics cannot be OR-ed across positions, so "touches this address" is two
sweeps: one where it is the sender, one where it is the recipient.
4. Log objects do NOT carry blockTimestamp, and native ETH emits no Transfer
log at all. Dates and ETH amounts each cost an extra call.
python fetcher.py 0xabc... # scan one address
python fetcher.py 0xabc... 0xdef... # scan a wallet set
"""
from __future__ import annotations
import json
import sys
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from typing import Any, Iterable
RPC_URL = "https://rpc.mainnet.chain.robinhood.com"
TRANSFER_TOPIC = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
# Cloudflare fronts this RPC and 403s Python's default agent. This is the
# smallest header set that gets through.
HEADERS = {
"content-type": "application/json",
"accept": "application/json",
"user-agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36"
),
}
# Robinhood Chain was stamped at block 1 on 2026-04-30 but crawled at up to
# 195s/block until it reached production cadence here. Scanning below this is
# half a million blocks of nothing.
LAUNCH_BLOCK = 505_859
BATCH_SIZE = 25 # measured: 25 calls in one round trip, ~900ms
RETRY_MAX = 6
RETRY_BASE = 1.5 # seconds; doubles each attempt
PACE_SECONDS = 0.35 # between heavy log queries
WINDOW_START = 4_000_000 # optimistic first window once the full sweep fails
WINDOW_MIN = 500 # at 18.9 logs/block chain-wide, 10k logs is ~530 blocks
WINDOW_MAX = 16_000_000
LOG_CAP = 10_000 # the node's hard ceiling per eth_getLogs query
MAX_BISECT_DEPTH = 40
class HttpStatus(Exception):
"""The node answered, with a status that is not success. retry_after is its
Retry-After header in seconds, when the transport could read one."""
def __init__(self, code: int, retry_after: float | None = None):
super().__init__(f"HTTP {code}")
self.code = code
self.retry_after = retry_after
class TransportError(Exception):
"""The request never got an answer at all: DNS, reset, timeout."""
class RpcError(Exception):
def __init__(self, message: str, code: int | None = None):
super().__init__(message)
self.code = code
def to_topic(address: str) -> str:
"""An indexed address topic is the address left-padded to 32 bytes."""
return "0x" + "0" * 24 + address.lower().removeprefix("0x")
def from_topic(topic: str) -> str:
return "0x" + topic[-40:]
@dataclass
class Leg:
token: str
frm: str
to: str
raw: int # base units; decimals are applied by the engine
log_index: int
@dataclass
class Tx:
tx: str
block: int
timestamp: int | None
sender: str | None
to: str | None
native_value: int # wei moved by the transaction itself
gas_wei: int # gasUsed x effectiveGasPrice
status: str | None
raw_legs: int | None # every Transfer in the tx, not only ours
legs: list[Leg] = field(default_factory=list)
class Fetcher:
def __init__(self, rpc: str = RPC_URL, timeout: int = 60, verbose: bool = False):
self.rpc = rpc
self.timeout = timeout
self.verbose = verbose
self.calls = 0
self.round_trips = 0
self.retries = 0
self.log_queries = 0
self.interpolated = 0
# ---------------------------------------------------------------- transport
def _post(self, payload: Any) -> Any:
"""POST with backoff.
Light calls are not throttled, but a real scan issues wide eth_getLogs
queries and those earn HTTP 429 quickly. Retrying with a widening pause
— and honouring Retry-After when the node sends one — is the difference
between a scan that finishes and one that dies halfway through a user's
history.
"""
data = json.dumps(payload).encode()
delay = RETRY_BASE
for attempt in range(RETRY_MAX):
self.round_trips += 1
try:
return json.loads(self._send(data))
except HttpStatus as e:
if e.code in (429, 500, 502, 503, 504) and attempt < RETRY_MAX - 1:
wait = max(delay, e.retry_after or 0)
self.retries += 1
if self.verbose:
print(f" HTTP {e.code}, waiting {wait:.1f}s "
f"(attempt {attempt + 1}/{RETRY_MAX})", file=sys.stderr)
self._sleep(wait)
delay *= 2
continue
raise RpcError(f"HTTP {e.code}", e.code) from None
except TransportError as e:
if attempt < RETRY_MAX - 1:
self.retries += 1
self._sleep(delay)
delay *= 2
continue
raise RpcError(f"transport: {e}") from None
raise RpcError("exhausted retries")
def _send(self, data: bytes) -> bytes:
"""One POST and nothing else: no retries, no pacing, no parsing.
This is the only method that knows which HTTP library is in use. It is
kept apart from the rules above so the tests can put a node in its
place -- one that answers 429, drops the connection, or names a
Retry-After -- and exercise those rules exactly as written.
"""
req = urllib.request.Request(self.rpc, data=data, headers=HEADERS)
try:
with urllib.request.urlopen(req, timeout=self.timeout) as r:
return r.read()
except urllib.error.HTTPError as e:
retry_after = None
hdr = e.headers.get("Retry-After") if e.headers else None
if hdr:
try:
retry_after = float(hdr)
except ValueError:
pass
raise HttpStatus(e.code, retry_after) from None
except (urllib.error.URLError, TimeoutError) as e:
raise TransportError(type(e).__name__) from None
def _sleep(self, seconds: float) -> None:
time.sleep(seconds)
def call(self, method: str, params: list) -> Any:
return self.batch([(method, params)])[0]
def batch(self, requests: list[tuple[str, list]]) -> list[Any]:
"""One round trip carries many calls; each error stays with its own."""
if not requests:
return []
self.calls += len(requests)
payload = [
{"jsonrpc": "2.0", "id": i, "method": m, "params": p}
for i, (m, p) in enumerate(requests)
]
rows = self._post(payload)
if isinstance(rows, dict):
rows = [rows]
out: list[Any] = [None] * len(requests)
for row in rows:
i = row.get("id", 0)
out[i] = RpcError(row["error"]["message"], row["error"].get("code")) \
if "error" in row else row.get("result")
return out
def batched(self, method: str, params_list: list[list], label: str = "") -> list[Any]:
out: list[Any] = []
for i in range(0, len(params_list), BATCH_SIZE):
chunk = params_list[i:i + BATCH_SIZE]
out.extend(self.batch([(method, p) for p in chunk]))
if self.verbose and label and sys.stderr.isatty():
done = min(i + len(chunk), len(params_list))
print(f" {label}: {done}/{len(params_list)}", end="\r", file=sys.stderr)
if self.verbose and label:
if sys.stderr.isatty():
print(" " * 44, end="\r", file=sys.stderr)
else:
print(f" {label}: {len(params_list)}", file=sys.stderr)
return out
# ---------------------------------------------------------------- reads
def head(self) -> int:
r = self.call("eth_blockNumber", [])
if isinstance(r, RpcError):
raise r
return int(r, 16)
def _try_logs(self, topics: list, start: int, end: int):
"""One eth_getLogs attempt. Returns the rows, or None if the node said
the window was too wide.
The node refuses in two different ways and both mean "narrow it":
"logs matched by query exceeds limit of 10000" when the result set is
too big, and "log query timed out" when the scan itself is too slow.
Neither is a silent truncation, which is the only reason paging is safe
at all — a node that quietly returned the first 10,000 rows would give
a confident, incomplete tax report.
"""
r = self.paced_logs({"fromBlock": hex(start), "toBlock": hex(end), "topics": topics})
if isinstance(r, RpcError):
msg = str(r).lower()
if any(t in msg for t in ("exceeds limit", "limit exceeded", "too many",
"response size", "timed out", "timeout",
"too large", "range")):
return None
raise r
if len(r) >= LOG_CAP:
return None # belt and braces against a silent cap
return r
def paced_logs(self, flt: dict):
"""eth_getLogs, spaced from the query before it.
Every log query in the tool comes through here, the price oracle's
included. Log queries are what the public node rate-limits, and the
oracle's are the heaviest the tool makes -- every WETH and USDG
transfer in a window. Pacing only the address sweeps left those
running back to back, and a full scan died of HTTP 429 at the pricing
stage with all of its transfers already read.
"""
if self.log_queries:
self._sleep(PACE_SECONDS)
self.log_queries += 1
return self.call("eth_getLogs", [flt])
def get_logs(self, topics: list, start: int, end: int) -> list[dict]:
"""Sweep a block range, widening the window while it holds and shrinking
it when the node pushes back.
Recursive bisection was the obvious approach and the wrong one: every
failure at the top of the tree costs a full query timeout before it
learns anything. Walking forward with an adaptive window pays that price
at most once.
"""
rows = self._try_logs(topics, start, end)
if rows is not None:
return rows # a light address needs one query
out: list[dict] = []
pos = start
window = WINDOW_START
while pos <= end:
stop = min(pos + window, end)
got = self._try_logs(topics, pos, stop)
if got is None:
if window <= WINDOW_MIN:
raise RpcError(
f"blocks {pos:,}-{stop:,} hold more than {LOG_CAP:,} matching "
"logs even at the minimum window; this address is too active "
"to page with the public RPC alone")
window = max(WINDOW_MIN, window // 4)
continue
out.extend(got)
pos = stop + 1
if self.verbose and sys.stderr.isatty():
pct = min(100, (pos - start) * 100 // max(1, end - start))
print(f" {pct:>3}% {len(out):,} logs window {window:,}",
end="\r", file=sys.stderr)
window = min(WINDOW_MAX, int(window * 1.8))
if self.verbose:
if sys.stderr.isatty():
print(" " * 60, end="\r", file=sys.stderr)
else:
print(f" paged {self.log_queries} queries, {len(out):,} logs",
file=sys.stderr)
return out
def transfers_for(self, addresses: Iterable[str], start: int = LAUNCH_BLOCK,
end: int | None = None) -> list[dict]:
"""Every Transfer leg touching these addresses.
Two sweeps, because topic positions cannot be OR-ed: position 1 is the
sender, position 2 is the recipient. A move between two of the user's
own wallets appears in both, so results are de-duplicated.
"""
end = self.head() if end is None else end
topics = [to_topic(a) for a in addresses]
seen: dict[str, dict] = {}
for position, label in ((1, "outgoing"), (2, "incoming")):
flt: list = [TRANSFER_TOPIC, None, None]
flt[position] = topics
logs = self.get_logs(flt, start, end)
if self.verbose:
print(f" {label}: {len(logs):,} logs", file=sys.stderr)
for log in logs:
seen[f"{log['transactionHash']}:{log['logIndex']}"] = log
return list(seen.values())
def block_times(self, block_nums: list[str], exact_limit: int = 400) -> dict[str, int]:
"""Block timestamps, exactly when that is cheap and by calibrated
interpolation when it is not.
Log objects do not carry blockTimestamp, so a date costs a call per
block. For a normal wallet that is a few dozen calls and worth paying.
For a very active one it dominates the whole scan, so instead we sample
anchors across the range and interpolate between them — safe here only
because the chain's cadence is both fast and regular. The anchors are
real, and the seconds-per-block is measured rather than assumed.
"""
uniq = sorted(set(block_nums), key=lambda b: int(b, 16))
if len(uniq) <= exact_limit:
rows = self.batched("eth_getBlockByNumber",
[[b, False] for b in uniq], "timestamps")
return {b: int(r["timestamp"], 16)
for b, r in zip(uniq, rows)
if isinstance(r, dict) and r.get("timestamp")}
anchors_n = min(64, len(uniq))
step = max(1, len(uniq) // anchors_n)
picks = uniq[::step]
if uniq[-1] not in picks:
picks.append(uniq[-1])
rows = self.batched("eth_getBlockByNumber",
[[b, False] for b in picks], "timestamp anchors")
anchors = sorted(
(int(b, 16), int(r["timestamp"], 16))
for b, r in zip(picks, rows) if isinstance(r, dict) and r.get("timestamp"))
if len(anchors) < 2:
return {b: t for b, t in ((b, a[1]) for b in uniq for a in anchors[:1])}
self.interpolated = len(uniq) - len(anchors)
out: dict[str, int] = {}
i = 0
for b in uniq:
n = int(b, 16)
while i < len(anchors) - 2 and anchors[i + 1][0] < n:
i += 1
(b0, t0), (b1, t1) = anchors[i], anchors[i + 1]
span = max(1, b1 - b0)
out[b] = int(t0 + (t1 - t0) * (n - b0) / span)
for b, t in ((hex(a[0]), a[1]) for a in anchors):
out[b] = t # anchors keep their exact value
return out
def hydrate(self, logs: list[dict]) -> list[Tx]:
"""Attach what the logs do not carry: dates, native ETH, gas."""
tx_hashes = list(dict.fromkeys(l["transactionHash"] for l in logs))
block_nums = list(dict.fromkeys(l["blockNumber"] for l in logs))
time_of = self.block_times(block_nums)
txs = self.batched("eth_getTransactionByHash",
[[h] for h in tx_hashes], "transactions")
receipts = self.batched("eth_getTransactionReceipt",
[[h] for h in tx_hashes], "receipts")
by_tx: dict[str, Tx] = {}
for h, tx, rc in zip(tx_hashes, txs, receipts):
tx = tx if isinstance(tx, dict) else None
rc = rc if isinstance(rc, dict) else None
gas = 0
if rc:
gas = int(rc.get("gasUsed", "0x0"), 16) * \
int(rc.get("effectiveGasPrice", "0x0"), 16)
raw_legs = None
if rc and isinstance(rc.get("logs"), list):
raw_legs = sum(1 for l in rc["logs"]
if l.get("topics") and l["topics"][0].lower() == TRANSFER_TOPIC)
by_tx[h] = Tx(
tx=h, block=0, timestamp=None,
# Who sent it decides whose gas it was, so the receipt's own
# "from" stands in when the transaction lookup came back empty.
sender=((tx or {}).get("from") or (rc or {}).get("from") or "").lower() or None,
to=(tx.get("to") or "").lower() or None if tx else None,
native_value=int(tx.get("value", "0x0"), 16) if tx else 0,
gas_wei=gas,
status=rc.get("status") if rc else None,
raw_legs=raw_legs,
)
for log in logs:
entry = by_tx.get(log["transactionHash"])
if entry is None:
continue
entry.block = int(log["blockNumber"], 16)
entry.timestamp = time_of.get(log["blockNumber"])
data = log.get("data") or "0x"
entry.legs.append(Leg(
token=log["address"].lower(),
frm=from_topic(log["topics"][1]),
to=from_topic(log["topics"][2]),
raw=int(data, 16) if data != "0x" else 0,
log_index=int(log["logIndex"], 16),
))
out = list(by_tx.values())
for t in out:
t.legs.sort(key=lambda l: l.log_index)
out.sort(key=lambda t: (t.block, t.tx))
return out
# ---------------------------------------------------------------- entry
def scan(self, addresses: list[str], start: int = LAUNCH_BLOCK,
end: int | None = None) -> dict:
began = time.time()
head = self.head() if end is None else end
logs = self.transfers_for(addresses, start, head)
transactions = self.hydrate(logs)
return {
"addresses": [a.lower() for a in addresses],
"headBlock": head,
"rawLogs": len(logs),
"transactions": transactions,
"rpcCalls": self.calls,
"roundTrips": self.round_trips,
"retries": self.retries,
"logQueries": self.log_queries,
"interpolatedTimestamps": self.interpolated,
"elapsed": round(time.time() - began, 2),
}
def main(argv: list[str]) -> int:
if len(argv) < 2:
print(__doc__.strip().split("\n\n")[-1])
return 1
addresses = [a for a in argv[1:] if a.startswith("0x") and len(a) == 42]
recent = None
for a in argv[1:]:
if a.startswith("--recent="):
recent = int(a.split("=", 1)[1])
if not addresses:
print("Give at least one 0x address (42 characters).", file=sys.stderr)
return 1
f = Fetcher(verbose=True)
print(f"scanning {len(addresses)} address(es) on Robinhood Chain\n", file=sys.stderr)
head = f.head()
begin = max(LAUNCH_BLOCK, head - recent) if recent else LAUNCH_BLOCK
if recent:
print(f"window: last {recent:,} blocks (from {begin:,})\n", file=sys.stderr)
result = f.scan(addresses, start=begin, end=head)
txs = result["transactions"]
print(f"head block {result['headBlock']:,}")
print(f"raw logs {result['rawLogs']:,}")
print(f"transactions {len(txs):,}")
print(f"rpc calls {result['rpcCalls']:,} in {result['roundTrips']:,} round trips")
print(f"log queries {result['logQueries']:,} ({result['retries']} retried)")
if result["interpolatedTimestamps"]:
print(f"timestamps {result['interpolatedTimestamps']:,} interpolated "
f"between measured anchors")
print(f"elapsed {result['elapsed']}s\n")
if txs:
with_value = sum(1 for t in txs if t.native_value > 0)
mine = {a.lower() for a in addresses}
gas_total = sum(t.gas_wei for t in txs if (t.sender or "") in mine)
multi = sum(1 for t in txs if (t.raw_legs or 0) > 2)
print(f"moving native ETH {with_value}/{len(txs)}")
print(f"more than 2 legs {multi}/{len(txs)}")
print(f"gas you sent {gas_total / 1e18:.10f} ETH\n")
print("first transactions:")
for t in txs[:6]:
when = time.strftime("%Y-%m-%d %H:%M", time.gmtime(t.timestamp)) if t.timestamp else "?"
print(f" {t.tx[:14]}… blk {t.block:>11,} {when} "
f"ours {len(t.legs)}/{t.raw_legs} legs "
f"value {t.native_value / 1e18:.6f} ETH")
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv))
# ---------------------------------------------------------------- token metadata
# The chain speaks in contract addresses and base units. Cost basis is written
# in symbols and decimal quantities. These four selectors bridge the two.
SEL_SYMBOL = "0x95d89b41"
SEL_DECIMALS = "0x313ce567"
SEL_NAME = "0x06fdde03"
def _decode_string(hexdata: str) -> str | None:
"""ERC-20 symbol() is usually a dynamic string, but older tokens return a
fixed bytes32. Both shapes appear in the wild, so handle both."""
if not hexdata or hexdata == "0x":
return None
raw = bytes.fromhex(hexdata[2:])
if len(raw) >= 64:
try:
offset = int.from_bytes(raw[0:32], "big")
length = int.from_bytes(raw[offset:offset + 32], "big")
if 0 < length <= 128:
text = raw[offset + 32: offset + 32 + length].decode("utf-8", "replace")
if text.strip():
return text.strip()
except (ValueError, IndexError):
pass
text = raw.rstrip(b"\x00").decode("utf-8", "replace").strip()
return text or None
def token_metadata(f: "Fetcher", tokens: list[str]) -> dict[str, dict]:
"""symbol / decimals / name for a set of contracts, in one batched pass.
These are constants, so reading them at the chain tip is safe even though
historical state is pruned.
"""
tokens = list(dict.fromkeys(t.lower() for t in tokens))
if not tokens:
return {}
calls = []
for t in tokens:
calls += [
[{"to": t, "data": SEL_SYMBOL}, "latest"],
[{"to": t, "data": SEL_DECIMALS}, "latest"],
[{"to": t, "data": SEL_NAME}, "latest"],
]
rows = f.batched("eth_call", calls, "token metadata")
out: dict[str, dict] = {}
for i, t in enumerate(tokens):
sym, dec, name = rows[i * 3], rows[i * 3 + 1], rows[i * 3 + 2]
decimals = None
if isinstance(dec, str) and dec not in ("0x", ""):
try:
decimals = int(dec, 16)
except ValueError:
decimals = None
if decimals is None or decimals > 36:
decimals = 18 # the overwhelming default
out[t] = {
"address": t,
"symbol": (_decode_string(sym) if isinstance(sym, str) else None) or t[:10],
"name": (_decode_string(name) if isinstance(name, str) else None) or "",
"decimals": decimals,
"resolved": isinstance(sym, str) and sym not in ("0x", ""),
}
return out
"""
Vrismcost — cost-basis engine.
Takes hydrated transactions from fetcher.py and walks them to a tax figure.
Pure logic: no network, no files, no demo data. The same code runs over a real
wallet and over the sample history the landing page shows.
The order of operations is the whole design:
1. Net every Transfer leg per asset, per transaction, across the user's whole
wallet set. On this chain a single trade emits 5.2 legs on average and one
real transaction carried 2,000 — router hops, fee sinks and mint/burn all
sit alongside the two that matter. Reading legs individually misreads the
majority of real traffic.
2. Classify the net, not the legs. A buy, a sale, a wallet move, a split and
an airdrop are five different tax outcomes that look alike in raw logs.
3. Price only what the chain states. A trade against the quote asset prices
itself exactly; anything else is queued for the owner, never guessed.
4. Walk the history once, in chain order. A lot exists from the transaction
that acquired it, so no sale reaches a lot bought after it. A corporate
action adjusts the lots open when it lands, exactly once, so a split after
a sale never rewrites the lots that sale consumed. And every matching
method keeps its own inventory: a second LIFO sale sees what the first
LIFO sale left, not what FIFO would have left.
Decimal throughout. Binary floats cannot represent money, and a tax figure
wrong by a cent is a wrong tax figure.
"""
from __future__ import annotations
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal, ROUND_HALF_UP, getcontext
from typing import Iterable
getcontext().prec = 50
D = Decimal
CENTS = D("0.01")
ZERO_ADDRESS = "0x" + "0" * 40
BURN_ADDRESSES = {ZERO_ADDRESS, "0x" + "0" * 36 + "dead"}
LONG_TERM_DAYS = 365
def money(x: D) -> str:
return str(x.quantize(CENTS, rounding=ROUND_HALF_UP))
def rate(x: D) -> str:
"""A per-unit price, not a money total. A token trading at four millionths
of a dollar is ordinary here, and cents precision would render it as zero."""
if x == 0:
return "0"
ax = abs(x)
places = 2 if ax >= 1 else (6 if ax >= D("0.0001") else 12)
return str(x.quantize(D(1).scaleb(-places), rounding=ROUND_HALF_UP).normalize())
def qty(x: D) -> str:
return format(x.normalize(), "f")
def units(raw: int, decimals: int) -> D:
return D(raw) / (D(10) ** decimals)
@dataclass
class Lot:
n: int
token: str
symbol: str
amount: D
unit_cost: D
block: int
acquired: datetime
tx: str
gas_quote: D
# None means untouched. Zero is a real state -- a lot sold down to nothing --
# and once doubled as "unset" here, which put every fully sold lot back on
# sale the next time a copy of the inventory was taken.
remaining: D | None = None
moves: list = field(default_factory=list)
corp: list = field(default_factory=list)
allocation: str | None = None
source: str = "chain" # chain, derived (ETH rate read off the chain) or user
def __post_init__(self):
if self.remaining is None:
self.remaining = self.amount
def copy(self) -> "Lot":
"""The same lot in another method's inventory, sharing nothing mutable."""
return Lot(n=self.n, token=self.token, symbol=self.symbol, amount=self.amount,
unit_cost=self.unit_cost, block=self.block, acquired=self.acquired,
tx=self.tx, gas_quote=self.gas_quote, remaining=self.remaining,
moves=list(self.moves), corp=list(self.corp),
allocation=self.allocation, source=self.source)
@property
def cost(self) -> D:
return self.amount * self.unit_cost
@dataclass
class Event:
tx: str
block: int
at: datetime
kind: str
taxable: bool | None
coverage: str # full | partial | none
gap: str | None
note: str
gained: dict[str, D]
given: dict[str, D]
gas_quote: D
native_out: D
raw_legs: int
corp: dict | None = None
answered: dict | None = None
pos: int = 0 # first log index: the order inside a block
gas_by_other: bool = False # sent, and so paid for, by an address not yours
gas_eth: D = D("0") # the gas this wallet paid, in ETH, before any rate
internal: dict = field(default_factory=dict) # asset -> quantity moved between your wallets
class Engine:
def __init__(self, wallets: Iterable[str], tokens: dict[str, dict],
quote_symbol: str = "USDG", native_symbol: str = "ETH",
native_price: D | None = None, answers: dict | None = None):
self.wallets = {w.lower() for w in wallets}
self.tokens = {k.lower(): v for k, v in tokens.items()}
self.quote = quote_symbol.upper()
self.native = native_symbol.upper()
# The ETH rate. Either a fixed Decimal, or anything exposing
# rate_at(block) — a PriceOracle reading it back off the chain. Left
# None, ETH legs stay unpriced and say so rather than being guessed.
self.native_price = native_price
# Answers the owner gave to the correction queue, keyed by tx hash and
# stamped user-supplied. They are the only figures in this engine that
# the chain did not prove, and they are never allowed to blend in.
self.answers = {k.lower(): v for k, v in (answers or {}).items()}
# ---------------------------------------------------------------- helpers
def symbol(self, token: str) -> str:
meta = self.tokens.get(token.lower())
return (meta or {}).get("symbol") or token[:10]
def decimals(self, token: str) -> int:
meta = self.tokens.get(token.lower())
return int((meta or {}).get("decimals", 18))
def native_rate(self, block: int) -> D | None:
"""The ETH rate at a block. A fixed figure applies everywhere; an
oracle is asked per block, because a rate that moved 30% across the
history is not one number."""
src = self.native_price
if src is None:
return None
if hasattr(src, "rate_at"):
return src.rate_at(block)
return src
def gas_in_quote(self, gas_wei: int, block: int = 0) -> D:
rate = self.native_rate(block)
if rate is None:
return D("0")
return (D(gas_wei) / (D(10) ** 18)) * rate
# ---------------------------------------------------------------- classify
def classify(self, tx) -> Event:
mine = self.wallets
burns, mints = [], []
delta: dict[str, D] = defaultdict(lambda: D("0"))
internal = []
for leg in tx.legs:
frm, to = leg.frm.lower(), leg.to.lower()
amount = units(leg.raw, self.decimals(leg.token))
sym = self.symbol(leg.token)
if to in BURN_ADDRESSES and frm in mine:
burns.append((sym, amount))
continue
if frm in BURN_ADDRESSES and to in mine:
mints.append((sym, amount))
continue
if frm in mine and to in mine:
internal.append((sym, amount))
continue
if to in mine:
delta[sym] += amount
elif frm in mine:
delta[sym] -= amount
native = units(tx.native_value, 18)
if native > 0 and (tx.sender or "").lower() in mine:
delta[self.native] -= native
gained = {k: v for k, v in delta.items() if v > 0}
given = {k: -v for k, v in delta.items() if v < 0}
moved: dict[str, D] = defaultdict(lambda: D("0"))
for sym, amount in internal:
moved[sym] += amount
moved = dict(moved)
at = datetime.fromtimestamp(tx.timestamp or 0, timezone.utc)
# Gas is paid by whoever sends the transaction. A bundler relaying a
# smart account, a counterparty sending you tokens, an issuer running a
# split or a distribution: each paid their own, and none of it is basis
# or selling cost of yours. Every transaction touching a wallet used to
# charge its whole gas to that wallet.
sent = (tx.sender or "").lower() in mine
gas_q = self.gas_in_quote(tx.gas_wei, tx.block) if sent else D("0")
common = dict(tx=tx.tx, block=tx.block, at=at, gas_quote=gas_q,
native_out=native, raw_legs=tx.raw_legs or len(tx.legs),
pos=min((l.log_index for l in tx.legs), default=0),
gas_by_other=bool(tx.gas_wei) and not sent,
gas_eth=units(tx.gas_wei, 18) if sent else D("0"),
internal=moved)
# Corporate actions first: they look like a disposal and are not one.
if burns and mints:
# A split reaches every wallet that holds the position, so one
# transaction can burn and mint in several of yours. Count them all.
bs, ms = burns[0][0], mints[0][0]
ba = sum((a for sym, a in burns if sym == bs), D("0"))
ma = sum((a for sym, a in mints if sym == ms), D("0"))
if bs == ms:
ratio = ma / ba if ba else D("1")
return Event(kind="SPLIT", taxable=False, coverage="full", gap=None,
gained={ms: ma}, given={bs: ba},
corp={"type": "split", "token": bs, "ratio": qty(ratio)},
note=f"{qty(ba)} {bs} burned and {qty(ma)} minted in one "
f"transaction — a {qty(ratio)}-for-1 split, not a sale.",
**common)
ratio = ma / ba if ba else D("1")
return Event(kind="MIGRATION", taxable=False, coverage="full", gap=None,
gained={ms: ma}, given={bs: ba},
corp={"type": "migration", "token": bs, "into": ms,
"ratio": qty(ratio)},
note=f"{bs} burned and {ms} minted together — a ticker "
"migration. Basis and acquisition dates carry across.",
**common)
if mints and not burns and not given:
ms, ma = mints[0]
if not gained and all(s == self.quote for s, _ in mints):
# The amount is stated exactly; what it was is not. A dividend, a
# bridge deposit and an issuance against the owner's own dollars
# all log a Transfer from 0x0, and the quote asset is recognised
# by a symbol any token can claim. So it is not booked as income
# on the chain's word: it is asked about, with every leg counted.
total = sum((a for _, a in mints), D("0"))
return Event(kind="DISTRIBUTION", taxable=True, coverage="none",
gap="not confirmed as income",
gained={self.quote: total}, given={},
corp={"type": "distribution", "token": self.quote,
"amount": qty(total)},
note=f"{qty(total)} {self.quote} minted with nothing paid. A "
"dividend, a bridge deposit and an issuance against your "
"own dollars all look like this. Vrismcost will not guess.",
**common)
return Event(kind="DISTRIBUTION", taxable=True, coverage="none",
gap="distribution has no stated value",
gained={ms: ma}, given={},
corp={"type": "distribution", "token": ms, "amount": qty(ma)},
note="Minted with nothing paid — a distribution. Income on "
"arrival, and it never joins the lot inventory.",
**common)
if not gained and not given:
if internal:
return Event(kind="INTERNAL", taxable=False, coverage="full", gap=None,
gained={}, given={},
note="Every leg nets to zero across your own wallets — "
"a move, not a disposal.", **common)
return Event(kind="UNRELATED", taxable=False, coverage="full", gap=None,
gained={}, given={},
note="No leg touches a registered wallet.", **common)
if gained and given:
if self.quote in given and self.quote not in gained:
return Event(kind="BUY", taxable=False, coverage="full", gap=None,
gained=gained, given=given,
note="Quote asset paid out — the trade prices itself exactly.",
**common)
if self.quote in gained and self.quote not in given:
return Event(kind="SELL", taxable=True, coverage="full", gap=None,
gained=gained, given=given,
note="Quote asset received — proceeds are exact.", **common)
if self.native in given and self.native not in gained:
if self.native_rate(tx.block) is not None:
return Event(kind="BUY", taxable=False, coverage="full", gap=None,
gained=gained, given=given,
note="Paid in native ETH, valued at the reference rate "
"you supplied. The quantity is exact; the rate is "
"yours.", **common)
return Event(kind="SWAP", taxable=True, coverage="partial",
gap="ETH leg has no quote price",
gained=gained, given=given,
note="Bought with native ETH. The quantity came from "
"tx.value, but ETH has no price in the logs.", **common)
if self.native in gained and self.native not in given:
if self.native_rate(tx.block) is not None:
return Event(kind="SELL", taxable=True, coverage="full", gap=None,
gained=gained, given=given,
note="Sold for native ETH, valued at the reference rate "
"you supplied.", **common)
return Event(kind="SWAP", taxable=True, coverage="partial",
gap="ETH leg has no quote price",
gained=gained, given=given,
note="Sold for native ETH. The quantity came from tx.value, "
"but ETH has no price in the logs.", **common)
return Event(kind="SWAP", taxable=True, coverage="none",
gap="no priceable leg", gained=gained, given=given,
note="Neither leg is the quote asset — nothing in this "
"transaction states a value.", **common)
if gained:
return Event(kind="RECEIVE", taxable=None, coverage="none",
gap="acquisition has no stated cost",
gained=gained, given={},
note="Assets arrived with nothing paid — income, a gift, or "
"your own unregistered wallet. Vrismcost will not guess.",
**common)
return Event(kind="SEND", taxable=None, coverage="none",
gap="disposal has no stated proceeds", gained={}, given=given,
note="Assets left with nothing received — a transfer out, a gift, "
"or an off-chain sale. Vrismcost will not guess.", **common)
# ---------------------------------------------------------------- answers
ANSWER_KINDS = {
"own": ("lot", False, "Your own wallet — the basis and date you gave it"),
"offchain": ("lot", False, "Bought off-chain — the basis you paid elsewhere"),
"income": ("lot", True, "Income on arrival, valued as you stated"),
"sold": ("proceeds", True, "Sold off-chain for the proceeds you stated"),
"manual": ("value", True, "Value you assigned to an otherwise unpriced trade"),
"eth_price": ("rate", True, "ETH rate you supplied for this transaction"),
"exclude": ("skip", False, "Excluded from the report at your instruction"),
}
def answer_for(self, tx: str):
a = self.answers.get(tx.lower())
if not a:
return None
effect = self.ANSWER_KINDS.get(a.get("action"))
if not effect:
return None
return {**a, "effect": effect[0], "taxable": effect[1], "label": effect[2]}
def native_source(self) -> str | None:
"""Where an ETH valuation came from: read off the chain, or supplied."""
if self.native_price is None:
return None
return "derived" if hasattr(self.native_price, "rate_at") else "user"
@staticmethod
def stated(answer: dict) -> D | None:
value = answer.get("value")
if value in (None, ""):
return None
try:
return D(str(value))
except Exception:
return None
@staticmethod
def stated_date(answer: dict, fallback: datetime) -> datetime:
when = answer.get("date")
if not when:
return fallback
try:
return datetime.strptime(str(when), "%Y-%m-%d").replace(tzinfo=timezone.utc)
except ValueError:
return fallback
def resolve(self, e: Event):
"""What one transaction puts into the books: the lots it opens and the
disposals it makes, each with a value and where that value came from.
Returns (buys, sells, status). A buy is (assets, total basis, gas,
source, acquired); a sell is (assets, total proceeds or None, gas,
source). status is "answered", "excluded" or None.
Every answer the queue offers does something here. An answer that
cannot be applied -- a value that is missing, a rate on a transaction
with no ETH leg -- resolves nothing, and the transaction stays open
rather than being counted as answered. And a transaction nobody has
priced still gives up the assets that left: they are gone from the
wallet, so they leave the inventory, unpriced."""
a = self.answer_for(e.tx)
e.answered = a
buys, sells = [], []
def assets(side):
return {s: v for s, v in side.items()
if v > 0 and s not in (self.quote, self.native)}
gained, given = assets(e.gained), assets(e.given)
if a:
effect, value = a["effect"], self.stated(a)
if effect == "skip":
return buys, sells, "excluded"
if effect == "lot" and not gained:
# Nothing arrived that needs a basis: income in the quote asset
# (its amount is its value), or your own wallet on the way out.
return buys, sells, "answered"
if effect == "lot" and value is not None:
buys.append((gained, value, D("0"), "user", self.stated_date(a, e.at)))
return buys, sells, "answered"
if effect == "value" and value is not None and (gained or given):
# One value, both ways: basis of what came in, proceeds of what left.
if gained:
buys.append((gained, value, D("0"), "user", e.at))
if given:
sells.append((given, value, e.gas_quote, "user"))
return buys, sells, "answered"
if effect == "proceeds" and value is not None and given:
sells.append((given, value, e.gas_quote, "user"))
return buys, sells, "answered"
if effect == "rate" and value is not None and value > 0:
gas = e.gas_eth * value
if self.native in e.given and gained:
buys.append((gained, e.given[self.native] * value + gas, gas, "user", e.at))
return buys, sells, "answered"
if self.native in e.gained and given:
sells.append((given, e.gained[self.native] * value, gas, "user"))
return buys, sells, "answered"
e.answered = None # nothing it could apply to
if e.kind == "BUY" and e.coverage == "full" and gained:
if self.quote in e.given:
paid, source = e.given[self.quote], "chain"
else:
paid, source = e.given[self.native] * self.native_rate(e.block), self.native_source()
buys.append((gained, paid + e.gas_quote, e.gas_quote, source, e.at))
elif e.kind == "SELL" and e.coverage == "full" and given:
if self.quote in e.gained:
proceeds, source = e.gained[self.quote], "chain"
else:
proceeds, source = e.gained[self.native] * self.native_rate(e.block), self.native_source()
sells.append((given, proceeds, e.gas_quote, source))
elif e.kind in ("SWAP", "SEND") and given:
sells.append((given, None, e.gas_quote, "UNRESOLVED"))
return buys, sells, None
# ---------------------------------------------------------------- lots
@staticmethod
def split_note(total: D | None, count: int) -> str | None:
"""One stated value covering several assets has to be divided, and
nothing on chain says how. Vrismcost divides evenly and records that it
did, rather than charging each asset the whole amount."""
if count <= 1:
return None
return (f"even split of {money(total)} across {count} assets" if total is not None
else f"one transaction, {count} assets")
def open_lots(self, e: Event, buys, n: int):
out = []
for assets, total, gas, source, acquired in buys:
count = D(len(assets))
note = self.split_note(total, len(assets))
for sym, amount in assets.items():
n += 1
out.append(Lot(n=n, token=sym, symbol=sym, amount=amount,
unit_cost=(total / count) / amount, block=e.block,
acquired=acquired, tx=e.tx, gas_quote=gas / count,
allocation=note, source=source or "chain"))
return out, n
def apply_corporate_action(self, e: Event, lots: list[Lot]) -> list[int]:
"""One split or migration, applied to the lots open when it lands.
Adjust quantity and cost per unit; never close a lot, never restart the
clock. Total cost is the invariant a split must preserve. Returns the
numbers of the lots it touched."""
ratio = D(e.corp["ratio"])
if ratio <= 0:
return []
touched = []
for l in lots:
if l.symbol != e.corp["token"] or l.remaining <= 0:
continue
before = l.amount * l.unit_cost
l.amount *= ratio
l.remaining *= ratio
l.unit_cost = before / l.amount
if e.corp["type"] == "migration":
l.symbol = l.token = e.corp["into"]
l.corp.append({"tx": e.tx, "type": e.corp["type"],
"ratio": e.corp["ratio"],
"at": e.at.date().isoformat()})
touched.append(l.n)
return touched
@staticmethod
def record_moves(e: Event, books: dict) -> None:
"""A move between your own wallets is not a disposal. The lots it
carried keep their basis and their dates, and the move is written onto
them, with its gas, oldest lots first."""
at = e.at.date().isoformat()
for sym, amount in e.internal.items():
for book in books.values():
left = amount
for l in sorted((l for l in book if l.symbol == sym and l.remaining > 0),
key=lambda l: (l.block, l.n)):
if left <= 0:
break
take = min(l.remaining, left)
l.moves.append({"tx": e.tx, "at": at, "qty": qty(take),
"gas": money(e.gas_quote)})
left -= take
# ---------------------------------------------------------------- matching
def match(self, method: str, lots: list[Lot], symbol: str, amount: D,
at: datetime, proceeds: D | None, gas: D) -> dict:
"""Match one disposal against one method's inventory.
proceeds None is a disposal nobody has priced: it still takes its lots,
and reports the basis it consumed with no gain. A quantity no lot covers
is reported as unmatched and earns no gain either -- selling what the
scanned history never bought is not a zero-cost sale, it is a gap."""
priced = proceeds is not None
pool = [l for l in lots if l.symbol == symbol and l.remaining > 0]
legs, basis, matched = [], D("0"), D("0")
if method == "AVERAGE" and pool:
total_amt = sum((l.remaining for l in pool), D("0"))
total_cost = sum((l.remaining * l.unit_cost for l in pool), D("0"))
unit = total_cost / total_amt
matched = min(amount, total_amt)
basis = unit * matched
frac = matched / amount if amount else D("0")
sources = {l.source for l in pool}
legs.append({"lot": 0, "acquired": "pooled", "qty": qty(matched),
"unitCost": rate(unit), "cost": money(basis), "costExact": str(basis),
"heldDays": None, "tx": "—",
"gasInBasis": money(sum((l.gas_quote * (l.remaining * matched / total_amt)
/ l.amount for l in pool if l.amount), D("0"))),
"gasShare": str(gas * frac),
"proceedsShare": str(proceeds * frac) if priced else None,
"lotSource": "user" if "user" in sources else
"derived" if "derived" in sources else "chain",
"movedWallets": any(l.moves for l in pool),
"split": any(l.corp for l in pool)})
elif pool:
if method == "FIFO":
pool.sort(key=lambda l: (l.block, l.n))
elif method == "LIFO":
pool.sort(key=lambda l: (-l.block, -l.n))
elif method == "HIFO":
pool.sort(key=lambda l: -l.unit_cost)
left = amount
for l in pool:
if left <= 0:
break
take = min(l.remaining, left)
cost = take * l.unit_cost
frac = take / amount
legs.append({"lot": l.n, "acquired": l.acquired.date().isoformat(),
"qty": qty(take), "unitCost": money(l.unit_cost),
"cost": money(cost), "costExact": str(cost),
"heldDays": (at - l.acquired).days, "tx": l.tx,
"gasInBasis": money(l.gas_quote * take / l.amount) if l.amount else "0.00",
"gasShare": str(gas * frac),
"proceedsShare": str(proceeds * frac) if priced else None,
"lotSource": l.source,
"movedWallets": bool(l.moves), "split": bool(l.corp)})
basis += cost
matched += take
left -= take
share = matched / amount if amount else D("0")
result = {"method": method, "legs": legs,
"proceeds": money(proceeds) if priced else None, "gas": money(gas),
"netProceeds": money(proceeds - gas) if priced else None,
"costBasis": money(basis),
"gain": money((proceeds - gas) * share - basis) if priced else None,
"unmatched": qty(max(D("0"), amount - matched)),
"effectiveUnitCost": rate(basis / matched) if matched else "0"}
if not pool:
result["note"] = "No acquired lot covers this disposal."
return result
def consume(self, method: str, lots: list[Lot], symbol: str, amount: D,
matched: dict) -> None:
"""Take a matched disposal out of that method's own inventory."""
if method == "AVERAGE":
# A pool sells at its average, so every lot in it shrinks by the
# same fraction and the average itself does not move.
pool = [l for l in lots if l.symbol == symbol and l.remaining > 0]
held = sum((l.remaining for l in pool), D("0"))
if held > 0:
keep = max(D("0"), held - amount) / held
for l in pool:
l.remaining *= keep
return
by_n = {l.n: l for l in lots}
for leg in matched["legs"]:
if leg["lot"] in by_n:
by_n[leg["lot"]].remaining -= D(leg["qty"])
def dispose(self, e: Event, books: dict, assets: dict, total: D | None,
gas: D, source: str) -> list[dict]:
count = D(len(assets))
note = self.split_note(total, len(assets))
out = []
for sym, amount in assets.items():
proceeds = None if total is None else total / count
share_gas = gas / count
methods = {}
for m in self.METHODS:
methods[m] = self.match(m, books[m], sym, amount, e.at, proceeds, share_gas)
self.consume(m, books[m], sym, amount, methods[m])
out.append({"tx": e.tx, "at": e.at.date().isoformat(), "kind": e.kind,
"symbol": sym, "qty": qty(amount),
"proceeds": None if proceeds is None else money(proceeds),
"gas": money(share_gas), "source": source, "allocation": note,
"unmatched": methods["FIFO"]["unmatched"], "methods": methods})
return out
@staticmethod
def lot_row(l: Lot) -> dict:
return {"n": l.n, "symbol": l.symbol, "qty": qty(l.amount),
"remaining": qty(l.remaining), "unitCost": rate(l.unit_cost),
"unitCostExact": str(l.unit_cost), "cost": money(l.cost),
"remainingCost": money(l.remaining * l.unit_cost),
"acquired": l.acquired.date().isoformat(),
"tx": l.tx, "gas": money(l.gas_quote), "split": l.corp, "moves": l.moves,
"allocation": l.allocation, "source": l.source}
# ---------------------------------------------------------------- report
METHODS = ("FIFO", "LIFO", "HIFO", "AVERAGE")
def run(self, transactions) -> dict:
events = [self.classify(t) for t in transactions]
# Block, then position inside it: a split and a sale in the same block
# still happen in the order the chain ran them.
events.sort(key=lambda e: (e.block, e.pos))
# One pass in chain order, one inventory per method. A lot joins when
# the transaction that acquired it lands, so a sale only reaches lots
# bought before it. A corporate action adjusts what is open at that
# moment, once. Each method consumes its own lots, so a second LIFO sale
# sees what the first LIFO sale left.
books = {m: [] for m in self.METHODS}
lots, disposals, corp, queue, excluded, unmatched = [], [], [], [], [], []
shapes: dict[str, int] = defaultdict(int)
cover: dict[str, int] = defaultdict(int)
answered_n = n = 0
for e in events:
shapes[e.kind] += 1
buys, sells, status = self.resolve(e)
if status == "excluded":
answered_n += 1
excluded.append({"tx": e.tx, "at": e.at.date().isoformat(),
"kind": e.kind, "label": e.answered["label"]})
elif status == "answered":
answered_n += 1
else:
cover[e.coverage] += 1
if e.coverage != "full":
queue.append(e)
new, n = self.open_lots(e, buys, n)
for l in new:
lots.append(l)
books["FIFO"].append(l) # FIFO's book is the report's lots
for m in self.METHODS[1:]:
books[m].append(l.copy())
if e.corp and e.corp["type"] in ("split", "migration"):
token = e.corp["token"]
held = [l for l in books["FIFO"] if l.symbol == token and l.remaining > 0]
qty_before = sum((l.remaining for l in held), D("0"))
cost_open = sum((l.remaining * l.unit_cost for l in held), D("0"))
touched = {m: self.apply_corporate_action(e, books[m])
for m in self.METHODS}["FIFO"]
if touched:
qty_after = sum((l.remaining for l in books["FIFO"] if l.n in touched), D("0"))
corp.append({"tx": e.tx, "at": e.at.date().isoformat(),
"type": e.corp["type"], "token": token,
"into": e.corp.get("into"), "ratio": e.corp["ratio"],
"lots": touched, "note": e.note,
# What was open when it landed, which is what it changed.
"qtyBefore": qty(qty_before), "qtyAfter": qty(qty_after),
"costOpen": money(cost_open)})
if e.kind == "INTERNAL" and e.internal:
self.record_moves(e, books)
for assets, total, gas, source in sells:
for d in self.dispose(e, books, assets, total, gas, source):
disposals.append(d)
# A priced disposal that the scanned history cannot cover is
# a gap, not a gain. An unpriced one is already held open.
if D(d["unmatched"]) > 0 and d["proceeds"] is not None:
unmatched.append({"tx": d["tx"], "at": d["at"], "symbol": d["symbol"],
"qty": d["unmatched"]})
total = len(events) or 1
realised = sum((D(d["methods"]["FIFO"]["gain"]) for d in disposals
if d["methods"]["FIFO"]["gain"] is not None), D("0"))
open_lots = [l for l in lots if l.remaining > 0]
blocked = cover["partial"] + cover["none"] + len(unmatched)
# Income the owner confirmed. An answered event leaves the queue, and the
# income sheet used to read only the queue, so a confirmed dividend left
# the export entirely while the export called itself complete.
income = []
for e in events:
a = e.answered
if not a or a.get("action") != "income" or not e.gained:
continue
stated = a.get("value")
defaulted = stated in (None, "") and set(e.gained) == {self.quote}
if defaulted:
stated = e.gained[self.quote] # an amount of the quote is its value
syms = list(e.gained)
income.append({
"tx": e.tx, "at": e.at.date().isoformat(),
"asset": ", ".join(syms),
"quantity": ", ".join(qty(e.gained[s]) for s in syms),
"value": money(D(str(stated))) if stated not in (None, "") else None,
"source": "user",
"note": "confirmed as income by you" + (" at the minted amount" if defaulted else ""),
})
return {
"wallets": sorted(self.wallets),
"quote": self.quote,
"transactions": len(events),
"shapes": dict(shapes),
"coverage": {"full": cover["full"], "answered": answered_n,
"partial": cover["partial"], "none": cover["none"],
"unmatched": len(unmatched),
"total": len(events),
"pct": round((cover["full"] + answered_n) / total * 100, 1),
"chainPct": round(cover["full"] / total * 100, 1),
"blocked": blocked},
"lots": [self.lot_row(l) for l in lots],
# The same lots as each method left them. `lots` is FIFO's; an
# export under another method carries that method's open lots.
"lotsByMethod": {m: [self.lot_row(l) for l in books[m]] for m in self.METHODS},
"openLotsByMethod": {m: sum(1 for l in books[m] if l.remaining > 0)
for m in self.METHODS},
"disposals": disposals,
"realised": money(realised),
"openLots": len(open_lots),
"gasPaidByOthers": sum(1 for e in events if e.gas_by_other),
"corporateActions": corp,
"income": income,
"answered": answered_n,
"excluded": excluded,
"unmatched": unmatched,
"queue": [{"tx": e.tx, "at": e.at.date().isoformat(), "kind": e.kind,
"answer": e.answered,
"coverage": e.coverage, "gap": e.gap, "note": e.note,
"given": {k: qty(v) for k, v in e.given.items()},
"gained": {k: qty(v) for k, v in e.gained.items()},
"rawLegs": e.raw_legs} for e in queue],
"events": events,
}
"""
Vrismcost — deriving the ETH price from the chain itself.
Cost basis needs a value for every ETH leg, and on this chain most real trades
settle in ETH rather than the stablecoin. Without a rate, coverage sits near
zero. With one, it jumps — so where the rate comes from decides whether the
whole report can be defended.
It does not come from a price feed. Historical state is pruned after about ten
minutes, so a pool's reserves at an old block cannot be read back; and an
external API would mean sending a user's blocks to a third party, which is the
one thing this tool promises not to do.
Logs survive, though, and that is the opening. Native ETH emits nothing, but
WETH is an ordinary ERC-20 and emits Transfer like anything else.
The first version of this read every transaction that happened to contain both
a WETH leg and a USDG leg, took the largest of each, and divided. That was
wrong, and the numbers said so: rates ranged from 180 to 62,000 USDG/ETH in a
single window. On a chain where a trade averages five legs, the biggest WETH
leg and the biggest USDG leg are usually different hops of a route — WETH into
one pool, USDG out of another, with some third token in between. Dividing them
compares two unrelated quantities.
A swap is anchored by its pool. One contract receives one token and sends the
other, inside the same transaction. So legs are matched by shared counterparty:
if WETH went to address P and USDG came out of P, those two amounts are the two
sides of one swap and their ratio is a real price. Each hop has its own pool, so
multi-hop routes stop contaminating the sample instead of having to be excluded.
What survives is then filtered on its own spread — median absolute deviation,
not min-max, because one manipulated swap should not be able to widen the band
enough to discredit an otherwise tight set.
Every rate returned carries its evidence: how many pools agreed, how far apart
they were, which transactions they came from. A rate you cannot audit is a
guess with better manners.
"""
from __future__ import annotations
from decimal import Decimal as D
from statistics import median
from fetcher import Fetcher, RpcError, TRANSFER_TOPIC
# Discovered by scanning the busiest contracts on chain 4663, not hardcoded
# from documentation. Verify these before trusting a report built on them.
WETH = "0x0bd7d308f8e1639fab988df18a8011f41eacad73"
USDG = "0x5fc5360d0400a0fd4f2af552add042d716f1d168"
WETH_DECIMALS = 18
USDG_DECIMALS = 6 # not 18 — getting this wrong moves the rate by 10^12
MIN_WETH = D("0.02") # ignore dust legs; their ratios are noise
MIN_USDG = D("20")
MAD_LIMIT = D("3.5") # reject a sample this many deviations out
# These are two of the busiest contracts on the chain: ~3,700 WETH legs and
# ~4,100 USDG legs per 600 blocks. A window of a few thousand blocks blows
# straight through the node's 10,000-log ceiling, so the default starts small
# and the search narrows on refusal rather than widening into it.
#
# How small was measured, not guessed (13 Sep 2026, three blocks across the
# chain's life). A half-width of 300 pulled 2.5-5.1 MB per attempt and kept
# 198-1,039 samples, to satisfy a minimum of three. A half-width of 25 pulled
# 0.18-0.39 MB and kept 11-65, with a median within 0.003-0.016% of the wide
# window's -- inside the spread either window reports. The wide default was
# paying thirteen times the bandwidth for no precision, and narrowing it halved
# a full-history run over a real wallet, 276s to 134s. A quiet window still
# widens by 3x until it has its three samples, exactly as before.
DEFAULT_WINDOW = 25
MIN_WINDOW = 5
MIN_SAMPLES = 3
# Swaps further apart than this are not a price. Measured medians are ~0.03%
# inside a 5-second window, with 0.1% at the 90th percentile; the worst window
# found (router legs matched as if they were pools) spread 76% and priced ETH
# a quarter low.
MAX_SPREAD = D("0.03")
def _amount(log: dict, decimals: int) -> D:
data = log.get("data") or "0x"
raw = int(data, 16) if data != "0x" else 0
return D(raw) / (D(10) ** decimals)
def _legs(f: Fetcher, address: str, lo: int, hi: int, decimals: int):
"""Every Transfer of one token in a range, keyed by transaction.
Returns None when the node refuses the window rather than an empty list.
Conflating "too many logs" with "no trades here" is what made the first
version report that the chain had no ETH price at all.
"""
r = f.paced_logs({"fromBlock": hex(lo), "toBlock": hex(hi),
"address": address, "topics": [TRANSFER_TOPIC]})
if isinstance(r, Exception) or len(r) >= 10_000:
return None
out: dict[str, list] = {}
for log in r:
t = log.get("topics") or []
if len(t) < 3:
continue
amount = _amount(log, decimals)
if amount <= 0:
continue
out.setdefault(log["transactionHash"], []).append(
("0x" + t[1][-40:].lower(), "0x" + t[2][-40:].lower(), amount))
return out
def sample_rates(f: Fetcher, centre: int, window: int = DEFAULT_WINDOW,
weth: str = WETH, usdg: str = USDG,
weth_decimals: int = WETH_DECIMALS,
usdg_decimals: int = USDG_DECIMALS):
"""Every WETH/USDG rate implied by a pool near `centre`.
None means the node refused the window; an empty list means it answered and
no pool traded the pair. The caller must be able to tell those apart.
"""
lo, hi = max(0, centre - window), centre + window
w = _legs(f, weth, lo, hi, weth_decimals)
if w is None:
return None
u = _legs(f, usdg, lo, hi, usdg_decimals)
if u is None:
return None
out = []
for tx, wl in w.items():
ul = u.get(tx)
if not ul:
continue
# Total each side moved per counterparty. A pool that took WETH and
# paid out USDG in the same transaction just quoted a price.
weth_in: dict[str, D] = {}
weth_out: dict[str, D] = {}
for frm, to, amt in wl:
weth_in[to] = weth_in.get(to, D(0)) + amt
weth_out[frm] = weth_out.get(frm, D(0)) + amt
usdg_in: dict[str, D] = {}
usdg_out: dict[str, D] = {}
for frm, to, amt in ul:
usdg_in[to] = usdg_in.get(to, D(0)) + amt
usdg_out[frm] = usdg_out.get(frm, D(0)) + amt
for pool in set(weth_in) | set(weth_out):
# WETH in, USDG out: the pool bought ETH.
wi, uo = weth_in.get(pool, D(0)), usdg_out.get(pool, D(0))
if wi >= MIN_WETH and uo >= MIN_USDG:
out.append({"tx": tx, "pool": pool, "weth": wi, "usdg": uo,
"rate": uo / wi, "side": "sold"})
continue
# USDG in, WETH out: the pool sold ETH.
wo, ui = weth_out.get(pool, D(0)), usdg_in.get(pool, D(0))
if wo >= MIN_WETH and ui >= MIN_USDG:
out.append({"tx": tx, "pool": pool, "weth": wo, "usdg": ui,
"rate": ui / wo, "side": "bought"})
return out
def _reject_outliers(samples):
"""Median absolute deviation, which one bad swap cannot widen.
A min-max band lets a single manipulated trade make a tight set look
worthless; MAD asks how far the typical sample sits from the middle.
"""
if len(samples) < 4:
return samples, 0
rates = sorted(s["rate"] for s in samples)
mid = D(str(median(rates)))
devs = sorted(abs(r - mid) for r in rates)
mad = D(str(median(devs)))
if mad == 0:
return samples, 0
kept = [s for s in samples if abs(s["rate"] - mid) <= MAD_LIMIT * mad]
return (kept, len(samples) - len(kept)) if len(kept) >= 3 else (samples, 0)
def derive_native_price(f: Fetcher, block: int, window: int = DEFAULT_WINDOW,
attempts: int = 7):
"""The ETH rate near a block, with the evidence behind it.
Narrows when the node refuses the window, widens when the window was quiet
or its swaps disagree. Returns None rather than a lonely number when the
chain has nothing consistent to say: an unpriced leg is an honest gap, a
fabricated rate is a wrong tax return.
"""
w = window
for _ in range(attempts):
samples = sample_rates(f, block, w)
if samples is None:
if w <= MIN_WINDOW:
return None
w = max(MIN_WINDOW, w // 3)
continue
if len(samples) >= MIN_SAMPLES:
kept, dropped = _reject_outliers(samples)
rates = sorted(s["rate"] for s in kept)
mid = D(str(median(rates)))
low, high = rates[0], rates[-1]
spread = (high - low) / mid if mid else D(0)
if spread > MAX_SPREAD:
w *= 3 # disagreement is not a price: look wider
continue
return {
"rate": mid,
"samples": len(kept),
"rejected": dropped,
"pools": len({s["pool"] for s in kept}),
"low": low,
"high": high,
"spreadPct": float(spread * 100),
"blockCentre": block,
"windowBlocks": w,
"confidence": "high" if spread < D("0.01") else "medium",
"evidence": [s["tx"] for s in kept[:5]],
"source": "WETH/USDG matched across a shared pool",
}
w *= 3
return None
class PriceOracle:
"""Caches derived rates per block bucket, so a scan of many transactions
does not re-query the same neighbourhood over and over."""
def __init__(self, fetcher: Fetcher, bucket: int = 200_000):
self.f = fetcher
self.bucket = bucket
self.cache: dict[int, dict | None] = {}
self.misses = 0
self.reasons: dict[int, str] = {}
def at(self, block: int) -> dict | None:
key = block // self.bucket
if key not in self.cache:
try:
got = derive_native_price(self.f, key * self.bucket + self.bucket // 2)
reason = None if got else "no agreeing pools"
except RpcError as e:
# An unpriced window is a state the report already carries
# honestly: its legs are exported as UNPRICED and listed. A scan
# that dies here throws away every transfer it has already
# proven. The transfer sweeps get no such leniency -- a missing
# transfer is a wrong report, not a gap.
got, reason = None, f"node unavailable ({e})"
self.cache[key] = got
if got is None:
self.misses += 1
self.reasons[key] = reason
return self.cache[key]
def rate_at(self, block: int) -> D | None:
got = self.at(block)
return got["rate"] if got else None
def summary(self) -> dict:
got = [v for v in self.cache.values() if v]
return {
"buckets": len(self.cache),
"resolved": len(got),
"unresolved": self.misses,
"unresolvedReasons": sorted(set(self.reasons.values())),
"bucketBlocks": self.bucket,
"rates": [{"block": v["blockCentre"], "rate": str(v["rate"]),
"samples": v["samples"], "confidence": v["confidence"],
"spreadPct": round(v["spreadPct"], 2)} for v in got],
}
"""
Vrismcost — the export.
The report is the product. Everything upstream exists so that one file can be
handed to an accountant and survive being questioned.
Four rules shape the format:
No comment lines in the data. A leading "# generated on …" offsets every
column the moment the file is opened in a spreadsheet, which is where these
files are always opened. Metadata goes in its own file.
Both hashes on every row. A leg is a claim about two moments — when the asset
was acquired and when it left. One hash proves half of that, so the schedule
carries the acquiring and the disposing transaction side by side.
Unresolved rows are exported, not dropped. A schedule that quietly omits what
the tool could not read looks complete and is not. Those rows carry UNPRICED
where a figure would be, and say why.
Provenance is a column, never a footnote. Every row is marked chain, derived
(valued at an ETH rate read off the chain), user (a figure the owner
supplied) or UNRESOLVED, so a reviewer can see at a glance which numbers the
blockchain proved and which ones someone typed.
Four files are written, because an accountant works in a spreadsheet and
separate sheets beat one file with sections:
<prefix>-disposals.csv the capital gains schedule, one row per leg
<prefix>-income.csv distributions, kept out of the lot inventory
<prefix>-open-lots.csv what is still held, carrying into the next year
<prefix>-summary.csv the metadata, including whether this is complete
"""
from __future__ import annotations
import csv
from decimal import Decimal as D
from pathlib import Path
UNPRICED = "UNPRICED"
DISPOSAL_COLUMNS = [
"asset", "quantity",
"acquired_date", "acquired_tx",
"disposed_date", "disposed_tx",
"cost_basis", "gas_in_basis",
"proceeds", "gas_on_disposal",
"gain", "days_held",
"method", "source", "note",
]
INCOME_COLUMNS = ["date", "asset", "quantity", "value", "source", "tx", "note"]
LOT_COLUMNS = ["lot", "asset", "quantity_remaining", "unit_cost", "cost_basis",
"acquired_date", "acquired_tx", "adjusted", "source"]
SUMMARY_COLUMNS = ["field", "value"]
def _plain(x) -> str:
"""A token quantity. Spreadsheet-safe: no separators, no scientific
notation. Full precision is kept, because 13,505,154.227261797 JEW is the
real quantity and rounding it would not balance against the chain."""
if x is None or x == "":
return ""
try:
return format(D(str(x)).normalize(), "f")
except Exception:
return str(x)
def _money(x) -> str:
"""A currency figure, always to the cent.
Decimal keeps every digit of a division, so a cost basis arrives as
2815.504805192172192644745473. That is not a defensible number to hand an
accountant: money has two decimal places, and a column that does not
respect that makes every other figure in the file look unconsidered."""
if x is None or x == "":
return ""
try:
return str(D(str(x)).quantize(D("0.01")))
except Exception:
return str(x)
def provenance(disposal_source: str, lot_source: str) -> str:
"""A row is only as proven as the least proven figure on it."""
if disposal_source == "UNRESOLVED":
return "UNRESOLVED"
sources = {disposal_source, lot_source}
if "user" in sources:
return "user"
if "derived" in sources:
return "derived"
return "chain"
def disposal_rows(report: dict, method: str = "FIFO") -> list[dict]:
"""One row per matched leg, one per quantity no lot covers, and one per
arrival the chain could not price.
Each leg carries its own share of the disposal's proceeds and gas, so the
gains down the sheet add up to the realised figure in the summary. A
disposal nobody has priced still shows the lots it consumed and their
basis, with UNPRICED where the proceeds and the gain would be."""
rows = []
disposed = set()
for d in report.get("disposals", []):
m = d["methods"].get(method)
if not m:
continue
disposed.add(d["tx"])
priced = d.get("proceeds") is not None
total_qty = D(d["qty"])
for leg in m["legs"]:
qty = D(leg["qty"])
cost = D(str(leg.get("costExact") or leg["cost"]))
gas = D(str(leg.get("gasShare") or "0"))
proceeds = D(str(leg["proceedsShare"])) if priced and leg.get("proceedsShare") is not None else None
notes = []
if d.get("allocation"):
notes.append(d["allocation"])
if leg.get("split"):
notes.append("lot adjusted by a corporate action")
if leg.get("movedWallets"):
notes.append("lot survived a wallet move")
rows.append({
"asset": d["symbol"],
"quantity": _plain(qty),
"acquired_date": leg["acquired"],
"acquired_tx": leg.get("tx", ""),
"disposed_date": d["at"],
"disposed_tx": d["tx"],
"cost_basis": _money(cost),
"gas_in_basis": _money(leg.get("gasInBasis", "0")),
"proceeds": _money(proceeds) if proceeds is not None else UNPRICED,
"gas_on_disposal": _money(gas),
"gain": _money(proceeds - gas - cost) if proceeds is not None else UNPRICED,
"days_held": "" if leg.get("heldDays") is None else leg["heldDays"],
"method": method,
"source": provenance(d.get("source", "chain"), leg.get("lotSource", "chain")),
"note": "; ".join(notes)[:180],
})
left = D(m.get("unmatched") or "0")
if left > 0 and total_qty > 0:
frac = left / total_qty
rows.append({
"asset": d["symbol"],
"quantity": _plain(left),
"acquired_date": "", "acquired_tx": "",
"disposed_date": d["at"], "disposed_tx": d["tx"],
"cost_basis": UNPRICED, "gas_in_basis": "",
"proceeds": _money(D(d["proceeds"]) * frac) if priced else UNPRICED,
"gas_on_disposal": _money(D(d.get("gas") or "0") * frac),
"gain": UNPRICED, "days_held": "", "method": method,
"source": "UNRESOLVED",
"note": "UNMATCHED: no acquisition in the scanned history covers this quantity",
})
# Arrivals the chain could not price still appear, marked. A distribution
# is income business and is kept on the income sheet, not here.
for q in report.get("queue", []):
if q.get("kind") == "DISTRIBUTION":
continue
answer = q.get("answer")
sides = [("gained", k, v) for k, v in (q.get("gained") or {}).items()]
if q.get("tx") not in disposed:
sides += [("given", k, v) for k, v in (q.get("given") or {}).items()]
for side, asset, amount in sides:
if asset in (report.get("quote"), "ETH") and side == "given":
continue
arriving = side == "gained"
rows.append({
"asset": asset,
"quantity": _plain(amount),
"acquired_date": q.get("at", "") if arriving else "",
"acquired_tx": q.get("tx", "") if arriving else "",
"disposed_date": "" if arriving else q.get("at", ""),
"disposed_tx": "" if arriving else q.get("tx", ""),
"cost_basis": UNPRICED,
"gas_in_basis": "",
"proceeds": "" if arriving else UNPRICED,
"gas_on_disposal": "",
"gain": UNPRICED,
"days_held": "",
"method": method,
"source": "user" if answer else "UNRESOLVED",
"note": (answer.get("label") if answer else q.get("note", ""))[:180],
})
return rows
def income_rows(report: dict) -> list[dict]:
"""Distributions are income on arrival. They never join the lot inventory
and they never offset a capital gain, so they get their own sheet."""
# Confirmed by the owner, at the value they gave -- or, for the quote asset,
# at the amount the chain states.
rows = [{"date": i.get("at", ""), "asset": i.get("asset", ""),
"quantity": _plain(i.get("quantity", "")),
"value": _money(i.get("value")) or UNPRICED,
"source": i.get("source", "user"), "tx": i.get("tx", ""),
"note": (i.get("note") or "")[:180]}
for i in report.get("income", [])]
# Still waiting on an answer.
for e in report.get("queue", []):
if e.get("kind") != "DISTRIBUTION":
continue
gained = e.get("gained") or {}
asset = next(iter(gained), "")
rows.append({
"date": e.get("at", ""),
"asset": asset,
"quantity": _plain(gained.get(asset, "")),
"value": _money((e.get("answer") or {}).get("value")) or UNPRICED,
"source": "user" if e.get("answer") else "UNRESOLVED",
"tx": e.get("tx", ""),
"note": e.get("note", "")[:180],
})
return sorted(rows, key=lambda r: r["date"])
def open_lot_rows(report: dict, method: str = "FIFO") -> list[dict]:
"""What is still held under the method this export is for. These carry
their basis and acquisition date into the next tax year rather than being
re-opened at whatever they are worth. What is left depends on the method:
after one sale from a cheap lot and a dear one, FIFO holds the dear lot and
LIFO the cheap one."""
rows = []
for l in (report.get("lotsByMethod") or {}).get(method, report.get("lots", [])):
remaining = D(l.get("remaining", l.get("qty", "0")))
if remaining <= 0:
continue
# The exact remaining cost, never a unit cost already rounded to cents
# multiplied back up: that drifted by up to half a unit per lot.
exact = l.get("remainingCost")
if exact is None:
exact = remaining * D(str(l.get("unitCostExact") or l.get("unitCost", "0")))
adjusted = [word for word, present in (("split", l.get("split")), ("moved", l.get("moves")))
if present]
rows.append({
"lot": l.get("n", ""),
"asset": l.get("symbol", ""),
"quantity_remaining": _plain(remaining),
"unit_cost": _plain(l.get("unitCost", "")),
"cost_basis": _money(exact),
"acquired_date": l.get("acquired", ""),
"acquired_tx": l.get("tx", ""),
"adjusted": "; ".join(adjusted),
"source": l.get("source") or ("user" if l.get("allocation") else "chain"),
})
return rows
def realised_under(report: dict, method: str) -> D:
"""Realised gain under the method this export is for.
The engine's own `realised` figure is FIFO, and says so wherever it is
printed. A sheet that names another method has to total that method's
gains, or it contradicts the disposal sheet sitting next to it.
"""
total = D("0")
for d in report.get("disposals", []):
gain = (d.get("methods", {}).get(method) or {}).get("gain")
if gain not in (None, ""): # an unpriced disposal has no gain to add
total += D(str(gain))
return total
def summary_rows(report: dict, meta: dict) -> list[dict]:
cov = report.get("coverage", {})
blocked = cov.get("blocked", 0)
excluded = report.get("excluded") or []
pairs = [
# The line a reviewer reads first, so it is the first line.
("status", "COMPLETE" if blocked == 0 else f"INCOMPLETE - {blocked} transaction(s) unresolved"),
("generated_by", "Vrismcost"),
("chain", f"{meta.get('chainName', 'Robinhood Chain')} ({meta.get('chainId', 4663)})"),
("wallets", "; ".join(report.get("wallets", []))),
("quote_asset", report.get("quote", "")),
("method", meta.get("method", "FIFO")),
("scanned_from_block", meta.get("fromBlock", "")),
("scanned_to_block", meta.get("toBlock", "")),
("transactions_read", report.get("transactions", 0)),
("priced_by_chain", cov.get("full", 0)),
("answered_by_owner", cov.get("answered", 0)),
("unresolved", blocked),
("unmatched_disposals", cov.get("unmatched", 0)),
("excluded_by_owner", len(excluded)),
("excluded_transactions", "; ".join(x["tx"] for x in excluded)),
("coverage_percent", cov.get("pct", 0)),
("eth_rate_source", meta.get("ethRateSource", "")),
("eth_rate_windows", meta.get("ethRateWindows", "")),
("realised_gain", _money(str(realised_under(report, meta.get("method", "FIFO"))))),
("open_lots", (report.get("openLotsByMethod") or {}).get(
meta.get("method", "FIFO"), report.get("openLots", 0))),
]
return [{"field": k, "value": "" if v is None else str(v)} for k, v in pairs]
def _write(path: Path, columns: list[str], rows: list[dict]) -> Path:
with open(path, "w", newline="", encoding="utf-8") as fh:
w = csv.DictWriter(fh, fieldnames=columns, extrasaction="ignore")
w.writeheader()
w.writerows(rows)
return path
def write_report(prefix: str, report: dict, meta: dict,
method: str = "FIFO") -> list[tuple[Path, int]]:
"""Write the four sheets. Returns each path with its row count."""
base = Path(prefix)
base.parent.mkdir(parents=True, exist_ok=True)
sets = [
(f"{base}-disposals.csv", DISPOSAL_COLUMNS, disposal_rows(report, method)),
(f"{base}-income.csv", INCOME_COLUMNS, income_rows(report)),
(f"{base}-open-lots.csv", LOT_COLUMNS, open_lot_rows(report, method)),
(f"{base}-summary.csv", SUMMARY_COLUMNS, summary_rows(report, {**meta, "method": method})),
]
return [(_write(Path(p), c, r), len(r)) for p, c, r in sets]
"""
Vrismcost — encrypted local vault.
Holds the part of a tax report the chain can never supply: which transfers were
your own wallet moves, what an untraceable acquisition actually cost, the ETH
rates you assigned, your wallet labels and notes. Answer the correction queue
once and it stays answered.
The file lives on your machine and nowhere else. There is no account, no sync,
no server that could be compelled or breached — losing the passphrase and the
recovery code means losing the contents, and that is the trade being made
deliberately.
Design:
master key AES-256, generated once from os.urandom
passphrase wrap scrypt (N=2^17, r=8, p=1) derives a key-encryption key
recovery wrap a second, independently generated 32-byte secret wraps the
same master key, so a forgotten passphrase is survivable
records AES-256-GCM, fresh 96-bit nonce per write
binding the vault id is passed as AEAD associated data, so a record
lifted from one vault will not decrypt into another
writes atomic: temp file, fsync, replace
scrypt at N=2^17 costs ~128MB and roughly a tenth of a second per attempt,
which is what makes a stolen vault expensive to attack offline. That is also
why unlocking is not instant.
Requires `cryptography` — the one dependency in this project. AES-GCM must come
from an audited implementation; composing one out of hashlib would be the kind
of cleverness that loses somebody their records.
"""
from __future__ import annotations
import base64
import json
import os
import secrets
import time
from dataclasses import dataclass
from hashlib import scrypt
from pathlib import Path
from typing import Any
try:
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
except ImportError: # pragma: no cover
raise SystemExit(
"Vrismcost needs the 'cryptography' package for the encrypted vault.\n"
" pip install cryptography"
)
FORMAT = "vrismcost-vault-1"
SCRYPT_N = 1 << 17
SCRYPT_R = 8
SCRYPT_P = 1
KEY_LEN = 32
NONCE_LEN = 12
RECOVERY_PREFIX = "VRSM"
DEFAULT_PATH = Path.home() / ".vrismcost" / "vault.json"
def b64(data: bytes) -> str:
return base64.b64encode(data).decode()
def unb64(text: str) -> bytes:
return base64.b64decode(text)
class VaultError(Exception):
pass
class WrongPassphrase(VaultError):
pass
def _derive(passphrase: str, salt: bytes) -> bytes:
return scrypt(passphrase.encode("utf-8"), salt=salt,
n=SCRYPT_N, r=SCRYPT_R, p=SCRYPT_P, dklen=KEY_LEN,
maxmem=256 * 1024 * 1024)
def _aad(vault_id: str, purpose: str) -> bytes:
"""Binds ciphertext to this vault and this role. A wrapped key cannot be
replayed as a record, and neither can cross vaults."""
return f"{FORMAT}|{vault_id}|{purpose}".encode()
def format_recovery(secret: bytes) -> str:
"""Grouped base32, easy to write down and read back without ambiguity."""
body = base64.b32encode(secret).decode().rstrip("=")
groups = [body[i:i + 5] for i in range(0, len(body), 5)]
return RECOVERY_PREFIX + "-" + "-".join(groups)
def parse_recovery(code: str) -> bytes:
raw = code.strip().upper().replace(" ", "").replace("-", "")
if raw.startswith(RECOVERY_PREFIX):
raw = raw[len(RECOVERY_PREFIX):]
pad = "=" * (-len(raw) % 8)
try:
secret = base64.b32decode(raw + pad)
except Exception:
raise VaultError("That recovery code is not readable.") from None
if len(secret) != KEY_LEN:
raise VaultError("That recovery code is the wrong length.")
return secret
def validate_passphrase(passphrase: str) -> None:
if len(passphrase) < 10:
raise VaultError("Use at least 10 characters. A vault backup can be "
"attacked offline for as long as someone likes.")
@dataclass
class Vault:
path: Path
meta: dict
key: bytes
data: dict
# ---------------------------------------------------------------- create
@staticmethod
def create(passphrase: str, path: Path = DEFAULT_PATH) -> tuple["Vault", str]:
validate_passphrase(passphrase)
path = Path(path)
if path.exists():
raise VaultError(f"A vault already exists at {path}")
vault_id = secrets.token_hex(16)
master = os.urandom(KEY_LEN)
salt = os.urandom(16)
kek = _derive(passphrase, salt)
wrapped_nonce = os.urandom(NONCE_LEN)
wrapped = AESGCM(kek).encrypt(wrapped_nonce, master, _aad(vault_id, "passphrase"))
recovery_secret = os.urandom(KEY_LEN)
rec_nonce = os.urandom(NONCE_LEN)
rec_wrapped = AESGCM(recovery_secret).encrypt(
rec_nonce, master, _aad(vault_id, "recovery"))
meta = {
"format": FORMAT,
"vaultId": vault_id,
"created": int(time.time()),
"revision": 0,
"kdf": {"name": "scrypt", "n": SCRYPT_N, "r": SCRYPT_R,
"p": SCRYPT_P, "salt": b64(salt)},
"wrappedKey": {"nonce": b64(wrapped_nonce), "ciphertext": b64(wrapped)},
"recovery": {"nonce": b64(rec_nonce), "ciphertext": b64(rec_wrapped)},
}
data = {"wallets": [], "answers": {}, "rates": {}, "settings": {},
"notes": {}}
v = Vault(path=path, meta=meta, key=master, data=data)
v.save()
return v, format_recovery(recovery_secret)
# ---------------------------------------------------------------- unlock
@staticmethod
def _read_envelope(path: Path) -> dict:
path = Path(path)
if not path.exists():
raise VaultError(f"No vault at {path}. Run `vrismcost vault init` first.")
env = json.loads(path.read_text(encoding="utf-8"))
if env.get("format") != FORMAT:
raise VaultError("That file is not a Vrismcost vault.")
return env
@staticmethod
def unlock(passphrase: str, path: Path = DEFAULT_PATH) -> "Vault":
env = Vault._read_envelope(path)
kdf = env["kdf"]
kek = scrypt(passphrase.encode("utf-8"), salt=unb64(kdf["salt"]),
n=kdf["n"], r=kdf["r"], p=kdf["p"], dklen=KEY_LEN,
maxmem=256 * 1024 * 1024)
try:
master = AESGCM(kek).decrypt(
unb64(env["wrappedKey"]["nonce"]),
unb64(env["wrappedKey"]["ciphertext"]),
_aad(env["vaultId"], "passphrase"))
except Exception:
raise WrongPassphrase("That passphrase does not unlock this vault.") from None
return Vault(path=Path(path), meta=env,
key=master, data=Vault._decrypt_body(env, master))
@staticmethod
def recover(recovery_code: str, new_passphrase: str,
path: Path = DEFAULT_PATH) -> tuple["Vault", str]:
"""Recovery rotates the code it consumed. A used code must not stay live."""
validate_passphrase(new_passphrase)
env = Vault._read_envelope(path)
secret = parse_recovery(recovery_code)
try:
master = AESGCM(secret).decrypt(
unb64(env["recovery"]["nonce"]),
unb64(env["recovery"]["ciphertext"]),
_aad(env["vaultId"], "recovery"))
except Exception:
raise VaultError("That recovery code does not match this vault.") from None
data = Vault._decrypt_body(env, master)
salt = os.urandom(16)
kek = _derive(new_passphrase, salt)
nonce = os.urandom(NONCE_LEN)
env["kdf"] = {"name": "scrypt", "n": SCRYPT_N, "r": SCRYPT_R,
"p": SCRYPT_P, "salt": b64(salt)}
env["wrappedKey"] = {
"nonce": b64(nonce),
"ciphertext": b64(AESGCM(kek).encrypt(
nonce, master, _aad(env["vaultId"], "passphrase"))),
}
next_secret = os.urandom(KEY_LEN)
rnonce = os.urandom(NONCE_LEN)
env["recovery"] = {
"nonce": b64(rnonce),
"ciphertext": b64(AESGCM(next_secret).encrypt(
rnonce, master, _aad(env["vaultId"], "recovery"))),
}
v = Vault(path=Path(path), meta=env, key=master, data=data)
v.save()
return v, format_recovery(next_secret)
@staticmethod
def _decrypt_body(env: dict, master: bytes) -> dict:
body = env.get("body")
if not body:
return {"wallets": [], "answers": {}, "rates": {}, "settings": {}, "notes": {}}
try:
plain = AESGCM(master).decrypt(
unb64(body["nonce"]), unb64(body["ciphertext"]),
_aad(env["vaultId"], "body"))
except Exception:
raise VaultError("The vault contents failed their integrity check. "
"The file has been altered or truncated.") from None
return json.loads(plain.decode("utf-8"))
# ---------------------------------------------------------------- persist
def save(self) -> None:
nonce = os.urandom(NONCE_LEN)
plain = json.dumps(self.data, separators=(",", ":"), sort_keys=True).encode()
self.meta["revision"] = int(self.meta.get("revision", 0)) + 1
self.meta["updated"] = int(time.time())
self.meta["body"] = {
"nonce": b64(nonce),
"ciphertext": b64(AESGCM(self.key).encrypt(
nonce, plain, _aad(self.meta["vaultId"], "body"))),
}
self.path.parent.mkdir(parents=True, exist_ok=True)
tmp = self.path.with_suffix(".tmp")
# Atomic: a crash mid-write must not leave a half vault behind.
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(self.meta, fh, indent=2)
fh.flush()
os.fsync(fh.fileno())
os.replace(tmp, self.path)
try:
os.chmod(self.path, 0o600)
except OSError:
pass # best effort; Windows ACLs do not map cleanly
def lock(self) -> None:
"""Drop the key from memory. Not a guarantee — Python strings and bytes
are not wipeable — but it stops casual reuse after a task finishes."""
self.key = b"\x00" * KEY_LEN
self.data = {}
# ---------------------------------------------------------------- content
def add_wallet(self, address: str, label: str = "") -> None:
address = address.lower()
for w in self.data["wallets"]:
if w["address"] == address:
w["label"] = label or w["label"]
return
self.data["wallets"].append({"address": address, "label": label})
def wallets(self) -> list[str]:
return [w["address"] for w in self.data["wallets"]]
def answer(self, tx: str, action: str, value: str | None = None,
note: str = "", date: str | None = None) -> None:
"""One answer to one queued transaction, stamped as user-supplied so the
export can always separate it from chain evidence. date, YYYY-MM-DD, is
when you really acquired what arrived -- a transfer from your own wallet
keeps the date you first bought it, not the day it moved."""
self.data["answers"][tx.lower()] = {
"action": action, "value": value, "note": note, "date": date,
"at": int(time.time()), "source": "user",
}
def unanswer(self, tx: str) -> None:
self.data["answers"].pop(tx.lower(), None)
def answers(self) -> dict[str, Any]:
return self.data["answers"]
def set_rate(self, symbol: str, date: str, value: str) -> None:
self.data["rates"].setdefault(symbol.upper(), {})[date] = value
def rate(self, symbol: str, date: str) -> str | None:
return self.data["rates"].get(symbol.upper(), {}).get(date)
def setting(self, key: str, value: Any = None) -> Any:
if value is None:
return self.data["settings"].get(key)
self.data["settings"][key] = value
return value
def summary(self) -> dict:
return {
"path": str(self.path),
"vaultId": self.meta["vaultId"],
"revision": self.meta["revision"],
"wallets": len(self.data["wallets"]),
"answers": len(self.data["answers"]),
"rates": sum(len(v) for v in self.data["rates"].values()),
"bytes": self.path.stat().st_size if self.path.exists() else 0,
}
"""
Vrismcost — the test suite.
A tax tool does not crash when it is wrong. It returns a number, confidently,
and the number is wrong. That is the failure mode these tests exist to catch:
every one of them pins a behaviour that would otherwise break silently.
Nothing here touches the network. The fetcher is exercised through its pure
functions and the engine through synthetic transactions, so the suite runs in
a few seconds, offline, and gives the same answer next year as it does today.
Tests that call a live chain are tests whose result depends on what strangers
traded this morning.
Several of these were written after the bug, not before it, and say so.
python test_vrismcost.py
python test_vrismcost.py -v
"""
from __future__ import annotations
import json
import tempfile
import unittest
from decimal import Decimal as D
from pathlib import Path
from fetcher import Leg, Tx, to_topic, from_topic, TRANSFER_TOPIC
from engine import Engine, money, qty, rate, units
import report as rep
from vault import (Vault, VaultError, WrongPassphrase,
format_recovery, parse_recovery, validate_passphrase)
# ---------------------------------------------------------------- fixtures
ME = "0x1111111111111111111111111111111111111111"
COLD = "0x2222222222222222222222222222222222222222"
POOL = "0x3333333333333333333333333333333333333333"
ROUTER = "0x4444444444444444444444444444444444444444"
FEE = "0x5555555555555555555555555555555555555555"
STRANGER = "0x6666666666666666666666666666666666666666"
ZERO = "0x" + "0" * 40
USDG = "0xaaaa000000000000000000000000000000000001"
NVDA = "0xaaaa000000000000000000000000000000000002"
TSLA = "0xaaaa000000000000000000000000000000000003"
TOKENS = {
USDG: {"symbol": "USDG", "decimals": 6, "name": "Global Dollar", "resolved": True},
NVDA: {"symbol": "NVDA", "decimals": 18, "name": "NVIDIA", "resolved": True},
TSLA: {"symbol": "TSLA", "decimals": 18, "name": "Tesla", "resolved": True},
}
DAY = 864_000 # blocks per day at 0.1s
def leg(token: str, frm: str, to: str, amount: str, i: int = 0) -> Leg:
decimals = TOKENS[token]["decimals"]
return Leg(token=token, frm=frm, to=to,
raw=int(D(amount) * (D(10) ** decimals)), log_index=i)
def tx(hash_: str, block: int, legs: list[Leg], *, native: str = "0",
gas: str = "0.0001", sender: str = ME, raw_legs: int | None = None) -> Tx:
return Tx(tx=hash_, block=block,
timestamp=1_782_000_000 + block // 10,
sender=sender, to=POOL,
native_value=int(D(native) * (D(10) ** 18)),
gas_wei=int(D(gas) * (D(10) ** 18)),
status="0x1", raw_legs=raw_legs if raw_legs is not None else len(legs),
legs=legs)
def engine(**kw) -> Engine:
opts = dict(wallets=[ME, COLD], tokens=TOKENS, quote_symbol="USDG")
opts.update(kw)
return Engine(**opts)
def buy(hash_: str, block: int, paid: str, got: str, **kw) -> Tx:
return tx(hash_, block, [leg(USDG, ME, POOL, paid, 0),
leg(NVDA, POOL, ME, got, 1)], **kw)
# ---------------------------------------------------------------- helpers
class Helpers(unittest.TestCase):
def test_topic_padding_round_trips(self):
self.assertEqual(to_topic(ME), "0x" + "0" * 24 + ME[2:])
self.assertEqual(from_topic(to_topic(ME)), ME)
def test_topic_is_case_insensitive(self):
self.assertEqual(to_topic(ME.upper()), to_topic(ME.lower()))
def test_transfer_topic_is_the_known_signature(self):
# keccak("Transfer(address,address,uint256)"). Hardcoding it is fine;
# silently mistyping it is not, and every scan depends on it.
self.assertEqual(
TRANSFER_TOPIC,
"0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")
def test_units_respects_token_decimals(self):
# USDG has 6 decimals, not 18. Getting this wrong moves every USDG
# figure by a factor of a trillion.
self.assertEqual(units(1_000_000, 6), D("1"))
self.assertEqual(units(10 ** 18, 18), D("1"))
def test_quantity_never_renders_in_scientific_notation(self):
# normalize() alone turns 70 into 7E+1, which is not a quantity anyone
# can paste into a tax return.
self.assertEqual(qty(D("70")), "70")
self.assertEqual(qty(D("0.00000001")), "0.00000001")
def test_rate_keeps_precision_small_prices_actually_have(self):
# A memecoin at four millionths of a dollar renders as 0.00 at cents
# precision, which reads as broken.
self.assertEqual(rate(D("0")), "0")
self.assertNotEqual(rate(D("0.0000048")), "0.00")
self.assertEqual(money(D("2815.504805192172")), "2815.50")
# ---------------------------------------------------------------- classify
class Classification(unittest.TestCase):
def test_quote_out_asset_in_is_a_buy(self):
e = engine()
ev = e.classify(buy("0xa", 100, "5296", "40"))
self.assertEqual(ev.kind, "BUY")
self.assertEqual(ev.coverage, "full")
self.assertFalse(ev.taxable)
def test_asset_out_quote_in_is_a_sale(self):
e = engine()
ev = e.classify(tx("0xb", 200, [leg(NVDA, ME, POOL, "70", 0),
leg(USDG, POOL, ME, "11781", 1)]))
self.assertEqual(ev.kind, "SELL")
self.assertTrue(ev.taxable)
def test_both_sides_mine_is_a_move_not_a_disposal(self):
# The single most expensive misread available: booking a wallet move as
# a sale invents a gain out of nothing.
e = engine()
ev = e.classify(tx("0xc", 300, [leg(NVDA, ME, COLD, "30")]))
self.assertEqual(ev.kind, "INTERNAL")
self.assertFalse(ev.taxable)
self.assertEqual(ev.gained, {})
self.assertEqual(ev.given, {})
def test_router_legs_net_away(self):
# A real trade here averages 5.2 legs. Reading them individually
# misreads the majority of traffic, so the net is what gets classified.
e = engine()
ev = e.classify(tx("0xd", 400, [
leg(USDG, ME, ROUTER, "2074.80", 0),
leg(USDG, ROUTER, POOL, "2069.61", 1),
leg(USDG, ROUTER, FEE, "5.19", 2),
leg(NVDA, POOL, ROUTER, "5", 3),
leg(NVDA, ROUTER, ME, "5", 4),
]))
self.assertEqual(ev.kind, "BUY")
self.assertEqual(ev.given, {"USDG": D("2074.80")})
self.assertEqual(ev.gained, {"NVDA": D("5")})
self.assertEqual(ev.raw_legs, 5)
def test_native_eth_comes_from_tx_value(self):
# ETH emits no Transfer log at all. One transaction in five moves some.
e = engine(native_price=D("3000"))
ev = e.classify(tx("0xe", 500, [leg(NVDA, POOL, ME, "2")], native="0.5"))
self.assertEqual(ev.kind, "BUY")
self.assertEqual(ev.given, {"ETH": D("0.5")})
def test_eth_leg_without_a_rate_is_partly_priced_not_guessed(self):
e = engine(native_price=None)
ev = e.classify(tx("0xf", 600, [leg(NVDA, POOL, ME, "2")], native="0.5"))
self.assertEqual(ev.kind, "SWAP")
self.assertEqual(ev.coverage, "partial")
self.assertIn("ETH", ev.gap)
def test_token_to_token_has_no_stated_value(self):
e = engine()
ev = e.classify(tx("0x10", 700, [leg(NVDA, ME, POOL, "10", 0),
leg(TSLA, POOL, ME, "5", 1)]))
self.assertEqual(ev.kind, "SWAP")
self.assertEqual(ev.coverage, "none")
def test_arrival_with_nothing_paid_is_not_guessed(self):
e = engine()
ev = e.classify(tx("0x11", 800, [leg(NVDA, STRANGER, ME, "12")]))
self.assertEqual(ev.kind, "RECEIVE")
self.assertIsNone(ev.taxable)
self.assertEqual(ev.coverage, "none")
def test_departure_with_nothing_received_is_not_guessed(self):
e = engine()
ev = e.classify(tx("0x12", 900, [leg(NVDA, ME, STRANGER, "12")]))
self.assertEqual(ev.kind, "SEND")
self.assertIsNone(ev.taxable)
def test_untouched_transaction_is_unrelated(self):
e = engine()
ev = e.classify(tx("0x13", 1000, [leg(NVDA, STRANGER, POOL, "5")],
sender=STRANGER))
self.assertEqual(ev.kind, "UNRELATED")
# ---------------------------------------------------------------- corporate
class CorporateActions(unittest.TestCase):
def split(self, block: int, before: str, after: str) -> Tx:
return tx("0xsplit", block, [leg(NVDA, ME, ZERO, before, 0),
leg(NVDA, ZERO, ME, after, 1)])
def test_burn_and_mint_of_one_token_is_a_split(self):
# This chain runs corporate actions as burn-and-mint, which to a
# log reader is a total disposal plus a free repurchase.
e = engine()
ev = e.classify(self.split(1000, "60", "240"))
self.assertEqual(ev.kind, "SPLIT")
self.assertFalse(ev.taxable)
self.assertEqual(ev.corp["ratio"], "4")
def test_burn_and_mint_of_different_tokens_is_a_migration(self):
e = engine()
ev = e.classify(tx("0xm", 1100, [leg(NVDA, ME, ZERO, "10", 0),
leg(TSLA, ZERO, ME, "10", 1)]))
self.assertEqual(ev.kind, "MIGRATION")
self.assertEqual(ev.corp["into"], "TSLA")
def test_mint_with_nothing_paid_is_income_not_a_lot(self):
e = engine()
ev = e.classify(tx("0xd1", 1200, [leg(USDG, ZERO, ME, "84.60")]))
self.assertEqual(ev.kind, "DISTRIBUTION")
self.assertTrue(ev.taxable)
def test_a_quote_asset_mint_asks_whether_it_was_income(self):
# The amount of a USDG mint is stated exactly; what it was is not. A
# dividend, a bridge deposit and an issuance against the owner's own
# dollars all log a Transfer from 0x0, and the quote is recognised by a
# symbol any token can claim. So it stays open, asked the right question.
ev = engine().classify(tx("0xd1", 1200, [leg(USDG, ZERO, ME, "84.60")]))
self.assertEqual((ev.kind, ev.coverage), ("DISTRIBUTION", "none"))
self.assertEqual(ev.gap, "not confirmed as income")
self.assertEqual(ev.gained, {"USDG": D("84.60")})
def test_a_quote_asset_mint_in_two_legs_states_the_whole_amount(self):
# Only the first minted leg was read. Harmless while nothing used the
# amount; wrong the moment an answer takes it as the value.
ev = engine().classify(tx("0xd2", 1200, [leg(USDG, ZERO, ME, "50", 0),
leg(USDG, ZERO, ME, "34.60", 1)]))
self.assertEqual(ev.gained, {"USDG": D("84.60")})
def test_a_mint_of_any_other_token_still_needs_a_value(self):
ev = engine().classify(tx("0xd3", 1200, [leg(NVDA, ZERO, ME, "2")]))
self.assertEqual((ev.kind, ev.coverage, ev.gap),
("DISTRIBUTION", "none", "distribution has no stated value"))
def test_split_preserves_total_cost_and_the_acquisition_date(self):
# The invariant a split must pass. If total cost moves, the adjustment
# was wrong; if the date moves, a holding period was reset on a
# position nobody sold.
e = engine()
r = e.run([buy("0xa", 1000, "5296", "40"), self.split(2000, "40", "160")])
lot = r["lots"][0]
self.assertEqual(lot["qty"], "160")
self.assertEqual(money(D(lot["cost"])), "5296.00")
self.assertTrue(lot["split"])
self.assertEqual(lot["acquired"], r["events"][0].at.date().isoformat())
def test_a_split_between_two_sales_lands_once(self):
# Written after the bug. The split was applied again at every sale and
# once more at the end, and before the sales that came first: one
# 4-for-1 split turned 40 shares into 2,560.
r = engine().run([buy("0xa", 1 * DAY, "5296", "40", gas="0"),
tx("0xs1", 10 * DAY, [leg(NVDA, ME, POOL, "10", 0),
leg(USDG, POOL, ME, "2000", 1)], gas="0"),
self.split(20 * DAY, "30", "120"),
tx("0xs2", 30 * DAY, [leg(NVDA, ME, POOL, "20", 0),
leg(USDG, POOL, ME, "1200", 1)], gas="0")])
first, second = (d["methods"]["FIFO"] for d in r["disposals"])
self.assertEqual(first["costBasis"], "1324.00") # 10 at 132.40, before the split
self.assertEqual(second["costBasis"], "662.00") # 20 at 33.10, after it
lot = r["lots"][0]
self.assertEqual((lot["qty"], lot["remaining"]), ("160", "100"))
self.assertEqual(len(lot["split"]), 1)
self.assertEqual(len(r["corporateActions"]), 1)
def test_a_split_leaves_lots_bought_after_it_alone(self):
r = engine().run([buy("0xa", 1 * DAY, "1000", "10", gas="0"),
self.split(5 * DAY, "10", "40"),
buy("0xb", 9 * DAY, "1000", "20", gas="0")])
before, after = r["lots"]
self.assertEqual(before["qty"], "40")
self.assertEqual(after["qty"], "20")
self.assertFalse(after["split"])
# ---------------------------------------------------------------- lots
class Lots(unittest.TestCase):
def test_acquisition_gas_joins_the_basis(self):
e = engine(native_price=D("2000"))
r = e.run([buy("0xa", 100, "1000", "10", gas="0.001")])
# 0.001 ETH at 2000 = 2.00 on top of 1000 paid.
self.assertEqual(money(D(r["lots"][0]["cost"])), "1002.00")
def test_nothing_unpriced_ever_becomes_a_zero_basis_lot(self):
# A zero-basis lot exports as 100% taxable gain. Queueing is the only
# honest alternative.
e = engine()
r = e.run([tx("0xr", 100, [leg(NVDA, STRANGER, ME, "12")])])
self.assertEqual(r["lots"], [])
self.assertEqual(len(r["queue"]), 1)
def test_a_wallet_move_does_not_close_the_lot(self):
e = engine()
r = e.run([buy("0xa", 100, "1000", "10"),
tx("0xmv", 200, [leg(NVDA, ME, COLD, "10")])])
self.assertEqual(len(r["lots"]), 1)
self.assertEqual(r["lots"][0]["remaining"], "10")
self.assertEqual(r["realised"], "0.00")
def test_gas_counts_only_when_your_wallet_sent_the_transaction(self):
# Written after the bug. A bundler's transaction, or a stranger's, had
# its whole gas charged to whichever wallet it touched.
e = engine(native_price=D("2000"))
r = e.run([buy("0xmine", 100, "1000", "10", gas="0.001"),
buy("0xrelayed", 200, "1000", "10", gas="0.01", sender=STRANGER)])
self.assertEqual([money(D(l["cost"])) for l in r["lots"]], ["1002.00", "1000.00"])
self.assertEqual(r["gasPaidByOthers"], 1)
# ---------------------------------------------------------------- matching
class Matching(unittest.TestCase):
def history(self):
"""Four buys at different prices, then one sale of 70."""
return [
buy("0xb1", 1 * DAY, "5296.00", "40", gas="0"),
buy("0xb2", 20 * DAY, "4156.25", "35", gas="0"),
buy("0xb3", 40 * DAY, "3852.50", "25", gas="0"),
buy("0xb4", 55 * DAY, "4248.00", "30", gas="0"),
tx("0xsell", 60 * DAY, [leg(NVDA, ME, POOL, "70", 0),
leg(USDG, POOL, ME, "11781.00", 1)], gas="0"),
]
def setUp(self):
self.r = engine().run(self.history())
self.m = self.r["disposals"][0]["methods"]
def test_fifo_takes_the_oldest_lots_first(self):
legs = self.m["FIFO"]["legs"]
self.assertEqual([l["qty"] for l in legs], ["40", "30"])
self.assertEqual(self.m["FIFO"]["costBasis"], "8858.50")
def test_lifo_takes_the_newest_lots_first(self):
self.assertEqual(self.m["LIFO"]["legs"][0]["qty"], "30")
self.assertEqual(self.m["LIFO"]["costBasis"], "9881.75")
def test_hifo_takes_the_dearest_lots_first(self):
legs = self.m["HIFO"]["legs"]
costs = [D(l["unitCost"]) for l in legs]
self.assertEqual(costs, sorted(costs, reverse=True))
self.assertEqual(self.m["HIFO"]["costBasis"], "10086.50")
def test_average_pools_every_open_lot(self):
self.assertEqual(len(self.m["AVERAGE"]["legs"]), 1)
self.assertEqual(self.m["AVERAGE"]["legs"][0]["acquired"], "pooled")
def test_every_method_sells_exactly_seventy(self):
for name, m in self.m.items():
with self.subTest(method=name):
total = sum(D(l["qty"]) for l in m["legs"])
self.assertEqual(total, D("70"), name)
self.assertEqual(D(m["unmatched"]), D("0"), name)
def test_the_method_chosen_changes_the_tax_owed(self):
# The whole argument for offering four of them.
gains = {k: D(v["gain"]) for k, v in self.m.items()}
self.assertEqual(max(gains.values()) - min(gains.values()), D("1228.00"))
self.assertLess(gains["HIFO"], gains["FIFO"])
def test_gain_is_always_proceeds_minus_basis(self):
for name, m in self.m.items():
with self.subTest(method=name):
self.assertEqual(
D(m["gain"]),
D(m["netProceeds"]) - D(m["costBasis"]), name)
def test_holding_days_are_measured_to_the_disposal(self):
legs = self.m["FIFO"]["legs"]
self.assertGreater(legs[0]["heldDays"], legs[1]["heldDays"])
def test_a_sale_cannot_reach_a_lot_bought_after_it(self):
# Written after the bug: every lot was open to every sale, including
# lots that did not exist yet when the sale happened.
r = engine().run([buy("0xb1", 1 * DAY, "1000", "10", gas="0"),
tx("0xs", 5 * DAY, [leg(NVDA, ME, POOL, "15", 0),
leg(USDG, POOL, ME, "3000", 1)], gas="0"),
buy("0xb2", 9 * DAY, "5000", "10", gas="0")])
fifo = r["disposals"][0]["methods"]["FIFO"]
self.assertEqual([(l["lot"], l["qty"]) for l in fifo["legs"]], [(1, "10")])
self.assertEqual(fifo["unmatched"], "5")
def test_every_method_sells_from_its_own_inventory(self):
# Written after the bug. A lot sold down to nothing came back for the
# next sale, and LIFO, HIFO and average all sold from FIFO's leftovers.
r = engine().run([buy("0xc1", 1 * DAY, "100", "1", gas="0"),
buy("0xc2", 2 * DAY, "300", "1", gas="0"),
tx("0xt1", 3 * DAY, [leg(NVDA, ME, POOL, "1", 0),
leg(USDG, POOL, ME, "400", 1)], gas="0"),
tx("0xt2", 4 * DAY, [leg(NVDA, ME, POOL, "1", 0),
leg(USDG, POOL, ME, "400", 1)], gas="0")])
bases = {m: [d["methods"][m]["costBasis"] for d in r["disposals"]]
for m in ("FIFO", "LIFO", "HIFO", "AVERAGE")}
self.assertEqual(bases, {"FIFO": ["100.00", "300.00"], "LIFO": ["300.00", "100.00"],
"HIFO": ["300.00", "100.00"], "AVERAGE": ["200.00", "200.00"]})
# ---------------------------------------------------------------- answers
class Answers(unittest.TestCase):
def test_an_answer_becomes_a_lot(self):
got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "12")])
e = engine(answers={"0xr": {"action": "offchain", "value": "2568.72"}})
r = e.run([got])
self.assertEqual(len(r["lots"]), 1)
self.assertEqual(money(D(r["lots"][0]["cost"])), "2568.72")
def test_one_value_across_two_assets_is_divided_not_duplicated(self):
# Found in testing, not in review: five answers of 125.00 produced ten
# lots of 125.00 each, doubling the cost basis, because the arriving
# transaction carried two tokens.
got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "10", 0),
leg(TSLA, STRANGER, ME, "5", 1)])
e = engine(answers={"0xr": {"action": "own", "value": "125.00"}})
r = e.run([got])
self.assertEqual(len(r["lots"]), 2)
total = sum(D(l["cost"]) for l in r["lots"])
self.assertEqual(money(total), "125.00")
self.assertTrue(all(l["allocation"] for l in r["lots"]))
def test_a_sole_asset_is_not_marked_as_split(self):
got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "10")])
e = engine(answers={"0xr": {"action": "own", "value": "100.00"}})
self.assertIsNone(e.run([got])["lots"][0]["allocation"])
def test_excluding_a_transaction_creates_nothing(self):
got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "10")])
e = engine(answers={"0xr": {"action": "exclude"}})
self.assertEqual(e.run([got])["lots"], [])
def test_answers_are_counted_apart_from_chain_evidence(self):
txs = [buy("0xa", 100, "1000", "10"),
tx("0xr", 200, [leg(NVDA, STRANGER, ME, "5")])]
plain = engine().run(txs)["coverage"]
answered = engine(answers={"0xr": {"action": "own", "value": "50"}}
).run(txs)["coverage"]
self.assertEqual(plain["full"], 1)
self.assertEqual(plain["blocked"], 1)
self.assertEqual(answered["full"], 1) # the chain proved no more
self.assertEqual(answered["answered"], 1) # the owner supplied one
self.assertEqual(answered["blocked"], 0)
def test_an_unanswered_quote_mint_holds_the_export_open(self):
r = engine().run([tx("0xd1", DAY, [leg(USDG, ZERO, ME, "84.60")], gas="0")])
self.assertEqual(r["coverage"]["blocked"], 1)
pairs = {x["field"]: x["value"] for x in rep.summary_rows(r, {})}
self.assertTrue(pairs["status"].startswith("INCOMPLETE"))
self.assertEqual([(x["value"], x["source"]) for x in rep.income_rows(r)],
[(rep.UNPRICED, "UNRESOLVED")])
def test_an_open_distribution_is_income_business_not_a_disposal(self):
# Written after the bug: the unanswered mint was written to the capital
# gains schedule as well as the income sheet.
r = engine().run([tx("0xd1", DAY, [leg(USDG, ZERO, ME, "84.60")], gas="0")])
self.assertEqual(rep.disposal_rows(r, "FIFO"), [])
def test_income_on_a_quote_mint_reaches_the_income_sheet_at_its_amount(self):
# Written after the bug. An `income` answer took the mint out of the
# queue, the income sheet read only the queue, and the export called
# itself COMPLETE with the income nowhere in it.
r = engine(answers={"0xd1": {"action": "income"}}).run(
[tx("0xd1", DAY, [leg(USDG, ZERO, ME, "84.60")], gas="0")])
self.assertEqual([(x["asset"], x["quantity"], x["value"], x["source"], x["tx"])
for x in rep.income_rows(r)],
[("USDG", "84.6", "84.60", "user", "0xd1")])
self.assertEqual((r["coverage"]["answered"], r["coverage"]["blocked"]), (1, 0))
self.assertEqual(r["lots"], [])
def test_income_stated_on_a_receive_reaches_the_income_sheet_and_the_lot(self):
# The same leak for any token: the lot was built, the income never written.
r = engine(answers={"0xr": {"action": "income", "value": "2747.40"}}).run(
[tx("0xr", DAY, [leg(NVDA, STRANGER, ME, "12")], gas="0")])
self.assertEqual(money(D(r["lots"][0]["cost"])), "2747.40")
self.assertEqual([(x["asset"], x["quantity"], x["value"], x["source"])
for x in rep.income_rows(r)],
[("NVDA", "12", "2747.40", "user")])
def test_own_or_exclude_on_a_quote_mint_is_not_income(self):
for action in ("own", "exclude"):
with self.subTest(action):
r = engine(answers={"0xd1": {"action": action}}).run(
[tx("0xd1", DAY, [leg(USDG, ZERO, ME, "84.60")], gas="0")])
self.assertEqual(rep.income_rows(r), [])
self.assertEqual(r["coverage"]["blocked"], 0)
# ---------------------------------------------------------------- export
class Export(unittest.TestCase):
def setUp(self):
self.r = engine().run([
buy("0xb1", 1 * DAY, "1000.00", "10", gas="0"),
tx("0xsell", 30 * DAY, [leg(NVDA, ME, POOL, "10", 0),
leg(USDG, POOL, ME, "1500.00", 1)], gas="0"),
tx("0xr", 40 * DAY, [leg(TSLA, STRANGER, ME, "5")]),
])
def test_money_is_quantised_and_quantities_are_not(self):
rows = rep.disposal_rows(self.r, "FIFO")
self.assertEqual(rows[0]["cost_basis"], "1000.00")
self.assertEqual(rows[0]["gain"], "500.00")
self.assertNotIn("E+", rows[0]["quantity"])
def test_both_transaction_hashes_are_on_the_row(self):
# One hash proves half a claim. A schedule needs the acquisition and
# the disposal side by side to be audit evidence.
row = rep.disposal_rows(self.r, "FIFO")[0]
self.assertEqual(row["acquired_tx"], "0xb1")
self.assertEqual(row["disposed_tx"], "0xsell")
def test_unresolved_rows_are_exported_not_dropped(self):
rows = rep.disposal_rows(self.r, "FIFO")
unresolved = [r for r in rows if r["source"] == "UNRESOLVED"]
self.assertEqual(len(unresolved), 1)
self.assertEqual(unresolved[0]["cost_basis"], rep.UNPRICED)
def test_the_summary_realised_gain_follows_the_chosen_method(self):
# Written after the bug. The summary sheet printed the FIFO gain under
# whatever method it named, so an export labelled HIFO disagreed with
# its own disposal sheet the moment two lots had different costs.
r = engine().run([
buy("0xcheap", 1 * DAY, "100.00", "1", gas="0"),
buy("0xdear", 2 * DAY, "300.00", "1", gas="0"),
tx("0xsell1", 30 * DAY, [leg(NVDA, ME, POOL, "1", 0),
leg(USDG, POOL, ME, "400.00", 1)], gas="0"),
])
for method, expected in (("FIFO", "300.00"), ("HIFO", "100.00"), ("LIFO", "100.00")):
pairs = {row["field"]: row["value"]
for row in rep.summary_rows(r, {"method": method})}
sheet_total = sum(D(row["gain"]) for row in rep.disposal_rows(r, method))
self.assertEqual(pairs["realised_gain"], expected, method)
self.assertEqual(D(pairs["realised_gain"]), sheet_total, method)
def test_the_summary_says_when_the_file_is_incomplete(self):
pairs = {r["field"]: r["value"] for r in rep.summary_rows(self.r, {})}
self.assertTrue(pairs["status"].startswith("INCOMPLETE"))
self.assertEqual(pairs["unresolved"], "1")
def test_a_clean_history_reports_complete(self):
clean = engine().run([buy("0xb1", DAY, "1000", "10", gas="0")])
pairs = {r["field"]: r["value"] for r in rep.summary_rows(clean, {})}
self.assertEqual(pairs["status"], "COMPLETE")
def test_open_lots_carry_their_basis_forward(self):
rows = rep.open_lot_rows(engine().run([buy("0xb", DAY, "1000", "10", gas="0")]))
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["cost_basis"], "1000.00")
def test_open_lots_follow_the_chosen_method(self):
r = engine().run([buy("0xcheap", 1 * DAY, "100.00", "1", gas="0"),
buy("0xdear", 2 * DAY, "300.00", "1", gas="0"),
tx("0xsell", 3 * DAY, [leg(NVDA, ME, POOL, "1", 0),
leg(USDG, POOL, ME, "400.00", 1)], gas="0")])
left = {m: [row["cost_basis"] for row in rep.open_lot_rows(r, m)]
for m in ("FIFO", "LIFO", "HIFO")}
self.assertEqual(left, {"FIFO": ["300.00"], "LIFO": ["100.00"], "HIFO": ["100.00"]})
def test_every_sheet_writes_its_header_even_when_empty(self):
with tempfile.TemporaryDirectory() as d:
written = rep.write_report(str(Path(d) / "t"), self.r, {}, "FIFO")
self.assertEqual(len(written), 4)
for path, _ in written:
first = path.read_text(encoding="utf-8").splitlines()[0]
self.assertTrue(first and "," in first, path.name)
self.assertFalse(first.startswith("#"), "a comment line offsets every column")
# ---------------------------------------------------------------- vault
class VaultTests(unittest.TestCase):
def setUp(self):
self.dir = tempfile.TemporaryDirectory()
self.path = Path(self.dir.name) / "vault.json"
def tearDown(self):
self.dir.cleanup()
def test_contents_round_trip_through_a_lock(self):
v, _ = Vault.create("a-long-enough-passphrase", self.path)
v.add_wallet(ME, "main")
v.answer("0xabc", "own", "100.00")
v.save()
v.lock()
again = Vault.unlock("a-long-enough-passphrase", self.path)
self.assertEqual(again.wallets(), [ME])
self.assertEqual(again.answers()["0xabc"]["value"], "100.00")
def test_nothing_readable_survives_on_disk(self):
v, _ = Vault.create("a-long-enough-passphrase", self.path)
v.add_wallet(ME, "my-main-wallet")
v.answer("0xabc", "own", "1234.56", "bought on an exchange")
v.save()
raw = self.path.read_text(encoding="utf-8")
for secret in (ME[2:10], "1234.56", "exchange", "my-main-wallet"):
self.assertNotIn(secret, raw)
def test_the_wrong_passphrase_is_refused(self):
Vault.create("a-long-enough-passphrase", self.path)
with self.assertRaises(WrongPassphrase):
Vault.unlock("not-the-passphrase", self.path)
def test_a_tampered_file_fails_its_integrity_check(self):
v, _ = Vault.create("a-long-enough-passphrase", self.path)
v.answer("0xabc", "own", "100")
v.save()
env = json.loads(self.path.read_text(encoding="utf-8"))
ct = env["body"]["ciphertext"]
env["body"]["ciphertext"] = ("B" if ct[0] != "B" else "C") + ct[1:]
self.path.write_text(json.dumps(env), encoding="utf-8")
with self.assertRaises(VaultError):
Vault.unlock("a-long-enough-passphrase", self.path)
def test_recovery_restores_and_retires_the_code_it_used(self):
v, code = Vault.create("a-long-enough-passphrase", self.path)
v.add_wallet(ME, "main")
v.save()
v.lock()
back, fresh = Vault.recover(code, "a-different-passphrase", self.path)
self.assertEqual(back.wallets(), [ME])
self.assertNotEqual(code, fresh)
with self.assertRaises(VaultError):
Vault.recover(code, "yet-another-passphrase", self.path)
def test_recovery_codes_round_trip(self):
secret = bytes(range(32))
self.assertEqual(parse_recovery(format_recovery(secret)), secret)
self.assertEqual(parse_recovery(format_recovery(secret).lower()), secret)
def test_short_passphrases_are_refused(self):
# A vault backup can be attacked offline for as long as someone likes.
with self.assertRaises(VaultError):
validate_passphrase("short")
# ---------------------------------------------------------------- the wire
# Everything below runs the real Fetcher and the real pipeline against a
# JSON-RPC node that lives in memory. Only _send is fake: the one method that
# knows HTTP exists.
import subprocess
import sys
from unittest import mock
import fetcher as fm
import price
import vrismcost
from fetcher import Fetcher, HttpStatus, RpcError, TransportError
WETH = price.WETH
HEAD = 3_000_000
BASE_TS = 1_782_000_000
def _word(n: int) -> str:
return n.to_bytes(32, "big").hex()
def _abi_string(text: str) -> str:
raw = text.encode()
return "0x" + _word(32) + _word(len(raw)) + raw.ljust((len(raw) + 31) // 32 * 32, b"\0").hex()
class FakeNode:
"""Enough of a Robinhood Chain node to run a scan end to end, offline."""
def __init__(self, tokens: dict):
self.tokens = tokens
self.logs: list[dict] = []
self.txs: dict[str, dict] = {}
self.receipts: dict[str, dict] = {}
self.requests: list[str] = []
def transfer(self, tx_hash, block, index, token, frm, to, amount):
decimals = self.tokens[token]["decimals"]
raw = int(D(amount) * (D(10) ** decimals))
log = {"address": token, "blockNumber": hex(block), "transactionHash": tx_hash,
"logIndex": hex(index), "data": "0x" + _word(raw),
"topics": [TRANSFER_TOPIC, to_topic(frm), to_topic(to)]}
self.logs.append(log)
self.txs[tx_hash] = {"from": ME, "to": POOL, "value": "0x0"}
rc = self.receipts.setdefault(tx_hash, {"gasUsed": "0x0", "effectiveGasPrice": "0x0",
"status": "0x1", "logs": []})
rc["logs"].append(log)
def answer(self, method, params):
self.requests.append(method)
if method == "eth_blockNumber":
return hex(HEAD)
if method == "eth_getLogs":
flt = params[0]
lo, hi = int(flt["fromBlock"], 16), int(flt["toBlock"], 16)
out = []
for log in self.logs:
if not lo <= int(log["blockNumber"], 16) <= hi:
continue
if flt.get("address") and log["address"] != flt["address"].lower():
continue
wanted = flt.get("topics") or []
if all(w is None or log["topics"][i] in (w if isinstance(w, list) else [w])
for i, w in enumerate(wanted)):
out.append(log)
return out
if method == "eth_getBlockByNumber":
return {"timestamp": hex(BASE_TS + int(params[0], 16) // 10)}
if method == "eth_getTransactionByHash":
return self.txs.get(params[0])
if method == "eth_getTransactionReceipt":
return self.receipts.get(params[0])
if method == "eth_call":
meta = self.tokens[params[0]["to"]]
return {fm.SEL_SYMBOL: _abi_string(meta["symbol"]),
fm.SEL_DECIMALS: "0x" + _word(meta["decimals"]),
fm.SEL_NAME: _abi_string(meta["name"])}[params[0]["data"]]
raise AssertionError(f"the fake node was asked for {method}")
def send(self, data: bytes) -> bytes:
payload = json.loads(data)
rows = payload if isinstance(payload, list) else [payload]
out = [{"jsonrpc": "2.0", "id": r["id"], "result": self.answer(r["method"], r["params"])}
for r in rows]
return json.dumps(out if isinstance(payload, list) else out[0]).encode()
def trade_node() -> FakeNode:
"""Buy 5 NVDA for 1,000 USDG, later sell 2 for 500."""
node = FakeNode(TOKENS)
node.transfer("0x" + "a1" * 32, 1_000_000, 0, USDG, ME, POOL, "1000")
node.transfer("0x" + "a1" * 32, 1_000_000, 1, NVDA, POOL, ME, "5")
node.transfer("0x" + "b2" * 32, 2_000_000, 0, NVDA, ME, POOL, "2")
node.transfer("0x" + "b2" * 32, 2_000_000, 1, USDG, POOL, ME, "500")
return node
class NodeFetcher(Fetcher):
"""The real Fetcher; only the wire and the clock are replaced."""
def __init__(self, node: FakeNode, **kw):
super().__init__(**kw)
self.node = node
self.slept: list[float] = []
def _send(self, data: bytes) -> bytes:
return self.node.send(data)
def _sleep(self, seconds: float) -> None:
self.slept.append(seconds)
class ScriptedFetcher(Fetcher):
"""A Fetcher whose wire replays a script of outcomes, one per attempt."""
def __init__(self, script):
super().__init__()
self.script = list(script)
self.slept: list[float] = []
def _send(self, data: bytes) -> bytes:
step = self.script.pop(0)
if isinstance(step, Exception):
raise step
return step
def _sleep(self, seconds: float) -> None:
self.slept.append(seconds)
OK_BODY = json.dumps([{"jsonrpc": "2.0", "id": 0, "result": "0x10"}]).encode()
class Transport(unittest.TestCase):
def test_rate_limit_is_retried_with_a_widening_pause(self):
f = ScriptedFetcher([HttpStatus(429), HttpStatus(429), OK_BODY])
self.assertEqual(f.call("eth_blockNumber", []), "0x10")
self.assertEqual(f.retries, 2)
self.assertEqual(f.slept, [fm.RETRY_BASE, fm.RETRY_BASE * 2])
def test_retry_after_longer_than_the_backoff_is_honoured(self):
f = ScriptedFetcher([HttpStatus(429, retry_after=10.0), OK_BODY])
f.call("eth_blockNumber", [])
self.assertEqual(f.slept, [10.0])
def test_a_client_error_is_not_retried(self):
f = ScriptedFetcher([HttpStatus(400)])
with self.assertRaises(RpcError) as caught:
f.call("eth_blockNumber", [])
self.assertEqual(caught.exception.code, 400)
self.assertEqual(f.slept, [])
def test_a_dead_network_gives_up_after_the_retry_budget(self):
f = ScriptedFetcher([TransportError("URLError")] * fm.RETRY_MAX)
with self.assertRaises(RpcError) as caught:
f.call("eth_blockNumber", [])
self.assertIn("transport", str(caught.exception))
self.assertEqual(len(f.slept), fm.RETRY_MAX - 1)
def test_every_log_query_is_paced_but_the_first(self):
# Written after the bug: the oracle's log queries went out back to
# back, and a full scan died of HTTP 429 at the pricing stage.
empty = json.dumps([{"jsonrpc": "2.0", "id": 0, "result": []}]).encode()
f = ScriptedFetcher([empty, empty, empty])
for _ in range(3):
f.paced_logs({"fromBlock": "0x1", "toBlock": "0x2"})
self.assertEqual(f.slept, [fm.PACE_SECONDS, fm.PACE_SECONDS])
class Oracle(unittest.TestCase):
def _swaps(self, rates):
"""A node holding one WETH->USDG swap per rate, all at one block."""
tokens = {price.WETH: {"symbol": "WETH", "decimals": 18, "name": "WETH"},
price.USDG: {"symbol": "USDG", "decimals": 6, "name": "USDG"}}
node = FakeNode(tokens)
for n, r in enumerate(rates):
h = "0x" + f"{n + 1:02x}" * 32
node.transfer(h, 1_000_000, 0, price.WETH, ME, POOL, "1")
node.transfer(h, 1_000_000, 1, price.USDG, POOL, ME, str(r))
return node
def test_swaps_that_agree_are_a_price(self):
got = price.derive_native_price(NodeFetcher(self._swaps([1880, 1881, 1882])), 1_000_000)
self.assertIsNotNone(got)
self.assertEqual(round(got["rate"]), 1881)
def test_swaps_that_disagree_are_not_a_price(self):
# Measured on chain, not invented: router legs matched as though they
# were pools priced ETH at 1,412 USDG with a 76% spread, where the
# neighbouring windows said about 1,890. The page promised that legs
# stay unpriced when the swaps disagree; the code used the number.
node = self._swaps([800, 940, 1880])
self.assertIsNone(price.derive_native_price(NodeFetcher(node), 1_000_000))
oracle = price.PriceOracle(NodeFetcher(node), bucket=100_000)
self.assertIsNone(oracle.rate_at(1_000_000))
self.assertEqual(oracle.summary()["unresolvedReasons"], ["no agreeing pools"])
def test_a_rate_limited_window_is_an_unpriced_gap_not_a_dead_scan(self):
# Written after the bug. The transfers were already read; losing them
# all because one price window was refused is the wrong trade.
f = ScriptedFetcher([HttpStatus(429)] * fm.RETRY_MAX)
oracle = price.PriceOracle(f, bucket=100_000)
self.assertIsNone(oracle.rate_at(1_234_567))
s = oracle.summary()
self.assertEqual((s["buckets"], s["resolved"], s["unresolved"]), (1, 0, 1))
self.assertTrue(any("node unavailable" in r for r in s["unresolvedReasons"]))
def test_a_quiet_window_says_so(self):
node = FakeNode(TOKENS)
oracle = price.PriceOracle(NodeFetcher(node), bucket=100_000)
self.assertIsNone(oracle.rate_at(1_234_567))
self.assertEqual(oracle.summary()["unresolvedReasons"], ["no agreeing pools"])
class Pipeline(unittest.TestCase):
def test_scanning_does_not_need_the_vault(self):
# The vault pulls in scrypt and AES-GCM. Reading a wallet needs neither,
# and should work where they are not installed.
code = "import sys; sys.modules['vault'] = None; import vrismcost"
r = subprocess.run([sys.executable, "-c", code], cwd=Path(__file__).parent,
capture_output=True, text=True)
self.assertEqual(r.returncode, 0, r.stderr)
def test_an_empty_range_returns_no_report(self):
result = vrismcost.run([ME], start=fm.LAUNCH_BLOCK, end=HEAD,
fetcher=NodeFetcher(FakeNode(TOKENS)))
self.assertIsNone(result["report"])
self.assertEqual(result["head"], HEAD)
def test_the_pipeline_reads_a_real_shaped_chain(self):
result = vrismcost.run([ME], start=fm.LAUNCH_BLOCK, end=HEAD, fetcher=NodeFetcher(trade_node()))
report = result["report"]
self.assertEqual(result["quote"], "USDG")
self.assertEqual(report["shapes"].get("BUY"), 1)
self.assertEqual(report["shapes"].get("SELL"), 1)
fifo = report["disposals"][0]["methods"]["FIFO"]
self.assertEqual(D(fifo["costBasis"]), D("400")) # 2 of 5 units bought for 1,000
self.assertEqual(D(fifo["gain"]), D("100"))
# ---------------------------------------------------------------- 0.2.6
# Written after the pre-release audit, before the fixes. Each of these was a
# way the export could be wrong while its summary said COMPLETE.
def swap(hash_: str, block: int, give: str, give_amt: str, get: str, get_amt: str, **kw) -> Tx:
return tx(hash_, block, [leg(give, ME, POOL, give_amt, 0), leg(get, POOL, ME, get_amt, 1)], **kw)
def summary(report: dict, meta: dict | None = None) -> dict:
return {row["field"]: row["value"] for row in rep.summary_rows(report, meta or {})}
class Resolutions(unittest.TestCase):
"""The queue offered answers the engine then ignored."""
def history(self):
return [buy("0xb", 1 * DAY, "1000", "10", gas="0"),
swap("0xsw", 5 * DAY, NVDA, "4", TSLA, "2", gas="0")]
def test_an_unanswered_swap_still_gives_up_the_lots_that_left(self):
r = engine().run(self.history())
self.assertEqual(r["lots"][0]["remaining"], "6")
d = r["disposals"][0]
self.assertIsNone(d["proceeds"])
self.assertEqual(d["methods"]["FIFO"]["costBasis"], "400.00")
self.assertIsNone(d["methods"]["FIFO"]["gain"])
self.assertTrue(summary(r)["status"].startswith("INCOMPLETE"))
row = [x for x in rep.disposal_rows(r, "FIFO") if x["disposed_tx"] == "0xsw"][0]
self.assertEqual((row["cost_basis"], row["proceeds"], row["gain"], row["source"]),
("400.00", rep.UNPRICED, rep.UNPRICED, "UNRESOLVED"))
def test_a_value_on_a_swap_is_proceeds_out_and_basis_in(self):
r = engine(answers={"0xsw": {"action": "manual", "value": "900"}}).run(self.history())
d = r["disposals"][0]
fifo = d["methods"]["FIFO"]
self.assertEqual((d["proceeds"], fifo["costBasis"], fifo["gain"]), ("900.00", "400.00", "500.00"))
tsla = [l for l in r["lots"] if l["symbol"] == "TSLA"][0]
self.assertEqual((money(D(tsla["cost"])), tsla["source"]), ("900.00", "user"))
self.assertEqual(r["coverage"]["blocked"], 0)
row = [x for x in rep.disposal_rows(r, "FIFO") if x["disposed_tx"] == "0xsw"][0]
self.assertEqual((row["gain"], row["source"]), ("500.00", "user"))
self.assertEqual(summary(r)["status"], "COMPLETE")
def test_sold_off_chain_is_a_disposal_at_the_stated_proceeds(self):
txs = [buy("0xb", 1 * DAY, "1000", "10", gas="0"),
tx("0xout", 5 * DAY, [leg(NVDA, ME, STRANGER, "10")], gas="0")]
r = engine(answers={"0xout": {"action": "sold", "value": "1500"}}).run(txs)
self.assertEqual(r["disposals"][0]["methods"]["FIFO"]["gain"], "500.00")
self.assertEqual(r["lots"][0]["remaining"], "0")
self.assertEqual(r["coverage"]["blocked"], 0)
self.assertEqual(rep.open_lot_rows(r), [])
def test_an_eth_rate_answer_prices_that_transaction(self):
paid = tx("0xeth", 5 * DAY, [leg(NVDA, POOL, ME, "2")], native="0.5", gas="0.001")
r = engine(answers={"0xeth": {"action": "eth_price", "value": "2000"}}).run([paid])
self.assertEqual(len(r["lots"]), 1)
self.assertEqual(money(D(r["lots"][0]["cost"])), "1002.00") # 0.5 ETH and 0.001 ETH gas at 2,000
self.assertEqual(r["coverage"]["blocked"], 0)
def test_an_answer_that_cannot_apply_leaves_the_transaction_open(self):
r = engine(answers={"0xsw": {"action": "eth_price", "value": "2000"}}).run(self.history())
self.assertEqual((r["coverage"]["answered"], r["coverage"]["blocked"]), (0, 1))
got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "12")])
r = engine(answers={"0xr": {"action": "offchain"}}).run([got])
self.assertEqual((r["lots"], r["coverage"]["blocked"]), ([], 1))
def test_an_exclusion_is_listed_in_the_export(self):
got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "12")])
r = engine(answers={"0xr": {"action": "exclude"}}).run([got])
self.assertEqual([x["tx"] for x in r["excluded"]], ["0xr"])
pairs = summary(r)
self.assertEqual((pairs["excluded_by_owner"], pairs["excluded_transactions"]), ("1", "0xr"))
self.assertEqual(pairs["status"], "COMPLETE")
def test_your_own_wallet_can_carry_its_original_date(self):
got = tx("0xr", 40 * DAY, [leg(NVDA, STRANGER, ME, "5")])
r = engine(answers={"0xr": {"action": "own", "value": "400", "date": "2026-07-01"}}).run([got])
lot = r["lots"][0]
self.assertEqual((lot["acquired"], lot["source"]), ("2026-07-01", "user"))
self.assertEqual(rep.open_lot_rows(r)[0]["source"], "user")
class Provenance(unittest.TestCase):
"""Every figure says where it came from: the chain, a rate derived from it, or you."""
def test_chain_evidence_is_marked_chain(self):
r = engine().run([buy("0xb", DAY, "1000", "10", gas="0"),
tx("0xs", 2 * DAY, [leg(NVDA, ME, POOL, "10", 0),
leg(USDG, POOL, ME, "1500", 1)], gas="0")])
self.assertEqual(rep.disposal_rows(r, "FIFO")[0]["source"], "chain")
def test_a_derived_eth_rate_is_marked_derived(self):
class Rate:
def rate_at(self, block):
return D("2000")
paid = tx("0xeth", 5 * DAY, [leg(NVDA, POOL, ME, "2")], native="0.5", gas="0")
sold = tx("0xs", 9 * DAY, [leg(NVDA, ME, POOL, "2", 0), leg(USDG, POOL, ME, "1500", 1)], gas="0")
r = engine(native_price=Rate()).run([paid, sold])
self.assertEqual(r["lots"][0]["source"], "derived")
self.assertEqual(rep.disposal_rows(r, "FIFO")[0]["source"], "derived")
class Schedule(unittest.TestCase):
def test_status_is_the_first_line_of_the_summary(self):
r = engine().run([buy("0xb", DAY, "1000", "10", gas="0")])
self.assertEqual(rep.summary_rows(r, {})[0]["field"], "status")
def test_a_sale_the_history_cannot_cover_holds_the_report_open(self):
# A scan bounded with --recent missed the purchase. The sale exported
# with a zero basis, its whole proceeds as gain, and the file said COMPLETE.
r = engine().run([tx("0xs", 5 * DAY, [leg(NVDA, ME, POOL, "10", 0),
leg(USDG, POOL, ME, "1500", 1)], gas="0")])
self.assertEqual(r["realised"], "0.00")
self.assertEqual(r["coverage"]["blocked"], 1)
rows = rep.disposal_rows(r, "FIFO")
self.assertEqual([(x["quantity"], x["source"], x["gain"]) for x in rows],
[("10", "UNRESOLVED", rep.UNPRICED)])
self.assertTrue(summary(r)["status"].startswith("INCOMPLETE"))
def test_the_schedule_adds_up_to_the_summary_with_gas(self):
# Every leg carried the whole disposal's gas, and no leg subtracted it.
r = engine(native_price=D("2000")).run([
buy("0xb1", 1 * DAY, "100.00", "1", gas="0.001"),
buy("0xb2", 2 * DAY, "300.00", "1", gas="0.001"),
tx("0xs", 9 * DAY, [leg(NVDA, ME, POOL, "2", 0),
leg(USDG, POOL, ME, "800.00", 1)], gas="0.002")])
rows = rep.disposal_rows(r, "FIFO")
self.assertEqual(sum(D(x["gain"]) for x in rows), D(r["realised"]))
self.assertEqual(sum(D(x["gas_on_disposal"]) for x in rows), D("4.00"))
self.assertEqual(D(summary(r)["realised_gain"]), D("392.00"))
def test_open_lots_carry_their_exact_cost(self):
# The sheet multiplied a unit cost already rounded to cents.
r = engine().run([buy("0xb", DAY, "1000", "3", gas="0"),
tx("0xs", 2 * DAY, [leg(NVDA, ME, POOL, "1", 0),
leg(USDG, POOL, ME, "400", 1)], gas="0")])
self.assertEqual(rep.open_lot_rows(r)[0]["cost_basis"], "666.67")
def test_a_wallet_move_is_recorded_on_the_lot_it_carried(self):
r = engine(native_price=D("2000")).run([
buy("0xa", 100, "1000", "10", gas="0"),
tx("0xmv", 200, [leg(NVDA, ME, COLD, "10")], gas="0.001"),
tx("0xs", 300, [leg(NVDA, COLD, POOL, "10", 0), leg(USDG, POOL, COLD, "1500", 1)],
gas="0", sender=COLD)])
self.assertEqual([(m["tx"], m["gas"]) for m in r["lots"][0]["moves"]], [("0xmv", "2.00")])
self.assertTrue(r["disposals"][0]["methods"]["FIFO"]["legs"][0]["movedWallets"])
self.assertIn("wallet move", rep.disposal_rows(r, "FIFO")[0]["note"])
def test_one_payment_for_two_assets_is_divided_not_duplicated(self):
r = engine().run([tx("0xb", DAY, [leg(USDG, ME, POOL, "1000", 0), leg(NVDA, POOL, ME, "10", 1),
leg(TSLA, POOL, ME, "5", 2)], gas="0")])
self.assertEqual(money(sum(D(l["cost"]) for l in r["lots"])), "1000.00")
class SplitAcrossWallets(unittest.TestCase):
def test_a_split_reaching_two_wallets_counts_both_and_states_what_was_open(self):
r = engine().run([
buy("0xa", 1 * DAY, "5000", "50", gas="0"),
tx("0xmv", 2 * DAY, [leg(NVDA, ME, COLD, "30")], gas="0"),
tx("0xsplit", 3 * DAY, [leg(NVDA, ME, ZERO, "20", 0), leg(NVDA, COLD, ZERO, "30", 1),
leg(NVDA, ZERO, ME, "80", 2), leg(NVDA, ZERO, COLD, "120", 3)])])
ca = r["corporateActions"][0]
self.assertEqual((ca["ratio"], ca["qtyBefore"], ca["qtyAfter"], ca["costOpen"]),
("4", "50", "200", "5000.00"))
self.assertIn("50 NVDA burned and 200 minted", ca["note"])
class Cli(unittest.TestCase):
def test_the_rpc_flag_reaches_the_pipeline(self):
seen = {}
def fake_run(addresses, **kw):
seen.update(kw)
return {"report": None}
with mock.patch.object(vrismcost, "run", fake_run):
vrismcost.main(["vrismcost.py", ME, "--rpc=http://127.0.0.1:8545"])
self.assertEqual(seen.get("rpc"), "http://127.0.0.1:8545")
def test_an_answer_keeps_the_date_you_give_it(self):
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "vault.json"
v, _ = Vault.create("a-long-enough-passphrase", path)
v.answer("0xabc", "own", "400", date="2026-07-01")
v.save()
v.lock()
again = Vault.unlock("a-long-enough-passphrase", path)
self.assertEqual(again.answers()["0xabc"]["date"], "2026-07-01")
if __name__ == "__main__":
unittest.main(verbosity=2)
# Vrismcost
Cost basis and realised P&L for **Robinhood Chain** (chain 4663), computed on
your machine.
You give it an address. It reads the chain, works out what each transaction
actually was, matches every disposal to the lots it consumed, and writes a
schedule your accountant can read. Anything it cannot prove goes in a queue
with the reason — it is never filled in with a plausible guess.
Version 0.2.6 · Python 3.10+ · one dependency
---
## What it contacts
One address, and nothing else:
https://rpc.mainnet.chain.robinhood.com
That is Robinhood Chain's public RPC. There is no account, no API key, no
analytics, no telemetry, no server of ours. Check it yourself before you run
anything — it takes one command:
```bash
grep -rn "https://" *.py | grep -v "^test_"
```
The only thing that ever leaves your computer is the address you asked about,
and it goes to that node. Your cost basis, your notes, your corrections and
your reports stay in a file on your disk.
**"Private" is not the same as anonymous.** The node can see which addresses
you looked up, and it can see your IP. If that matters to you, use a VPN or run
your own node and point Vrismcost at it with `--rpc=URL`.
---
## Install
```bash
pip install -r requirements.txt
python test_vrismcost.py
```
The test suite runs offline in a few seconds. If it is not green, do not trust
the numbers.
`cryptography` is the only dependency, and only the encrypted vault needs it.
AES-GCM has to come from an audited implementation; composing one out of
`hashlib` would be the kind of cleverness that loses somebody their records.
---
## Use
```bash
# one wallet, full history
python vrismcost.py 0xabc...
# a wallet set, netted together — moves between them are not disposals
python vrismcost.py 0xabc... 0xdef...
# write the schedule
python vrismcost.py 0xabc... --csv=2026 --method=HIFO
```
| Flag | What it does |
| --- | --- |
| `--recent=N` | Only scan the last N blocks. A sale of anything bought before that window has no lot to match, so it is marked `UNMATCHED` and the report stays INCOMPLETE. |
| `--method=` | `FIFO` (default), `LIFO`, `HIFO`, `AVERAGE` |
| `--quote=` | Name the quote asset. Detected from your tokens otherwise. |
| `--eth=3455.20` | Fix the ETH rate yourself instead of deriving it |
| `--eth=off` | Leave ETH legs unpriced |
| `--csv=PREFIX` | Write the four sheets |
| `--json=FILE` | Write the whole report as JSON |
| `--vault` | Use the wallets and answers in your vault |
| `--rpc=URL` | Read from another node, such as your own, instead of the public RPC |
### The vault
The chain cannot tell you which transfers were your own wallet moves, what an
untraceable acquisition cost, or whether an arrival was income or a gift. You
answer once, and the answer is kept — encrypted, on your machine.
```bash
python vrismcost.py vault init # writes ~/.vrismcost/vault.json
python vrismcost.py vault add 0xabc... main
python vrismcost.py vault answer 0xTX own 2568.72 2026-07-01
python vrismcost.py --vault
```
| Answer | What it does |
| --- | --- |
| `own VALUE [DATE]` | It came from a wallet of yours: a lot at that basis, dated DATE (YYYY-MM-DD) or, without one, the day it arrived |
| `offchain VALUE [DATE]` | You bought it elsewhere: a lot at the basis you paid |
| `income VALUE` | Income on arrival, on the income sheet. A token also becomes a lot at that value; USDG needs no value |
| `sold VALUE` | It left and was sold off-chain: a disposal at those proceeds, matched against your lots |
| `manual VALUE` | An unpriced trade: the proceeds of what left and the basis of what arrived |
| `eth_price RATE` | The ETH rate for that one transaction |
| `exclude` | Left out of the figures, and listed in the summary so the omission is visible |
An answer that cannot apply — a value that is missing, an ETH rate on a
transaction with no ETH leg — answers nothing, and the transaction stays open.
Write the recovery code down on paper. It is not stored anywhere and it is
shown once. Lose both it and the passphrase and the contents are gone — that
trade is deliberate, and it is the same trade that means nobody else can read
the file either.
---
## What comes out
Four sheets, because an accountant works in a spreadsheet:
| File | What it holds |
| --- | --- |
| `PREFIX-disposals.csv` | The capital-gains schedule, one row per matched leg |
| `PREFIX-income.csv` | Distributions, kept out of the lot inventory |
| `PREFIX-open-lots.csv` | What is still held, with its basis and dates |
| `PREFIX-summary.csv` | Whether any of it is complete, on the first line, then the metadata |
Every row carries **both** transaction hashes, the acquiring and the disposing
one, because a leg is a claim about two moments and one hash proves half of it.
Every row carries a `source` column: `chain` for figures the blockchain
states, `derived` for values at an ETH rate read off the chain, `user` for
anything you supplied, and `UNRESOLVED` for what is still open. A row is only as
proven as the least proven figure on it.
Each leg carries its own share of the disposal's proceeds and gas, so the gains
down the sheet add up to the realised gain in the summary.
Rows it could not price are exported marked `UNPRICED`, not dropped. A schedule
that quietly omits what the tool could not read looks complete and is not. While
anything is open, the summary's first line says INCOMPLETE.
---
## How the numbers are found
**A trade is a transaction, not a log.** On this chain one trade emits 5.2
Transfer logs on average and 70.9% emit more than two — router hops, fee sinks,
mint and burn. One real transaction carried 2,000 legs, of which one touched
the wallet being scanned. Legs are netted per asset per transaction, and the
net is what gets classified.
**Native ETH emits no log at all.** One transaction in five moves some. The
amount comes from the transaction's own `value` field, which is the only record
there is.
**Prices come out of the chain, not a feed.** A trade against USDG states its
own price exactly. Most real trades here settle in ETH, so the ETH rate is
derived: WETH and USDG legs matched **across a shared pool** — one contract
taking one token and paying the other inside a single transaction — then the
median of the matched swaps. Inside a five-second window they agree to 0.03% at
the median and within 0.1% nine times in ten (measured 12–13 September 2026). A
window whose swaps disagree by more than 3% leaves those legs unpriced rather
than filled in.
Historical state is pruned after about ten minutes on this chain, so a pool's
reserves at an old block cannot be read back. Logs survive; that is why the
rate is reconstructed from them.
**A USDG mint is asked about, not booked.** Its amount is exact, but a dividend,
a bridge deposit and an issuance against your own dollars all log the same
Transfer from the zero address, and the quote asset is recognised by its symbol,
which any token can claim. So it waits in the queue as *not confirmed as income*.
Answer `income` and it is booked on the income sheet at the minted amount; answer
`own` and it is not income at all.
**What left is gone, priced or not.** When an asset leaves in a transaction
nothing prices — a token-for-token swap, a transfer out — it still leaves your
inventory. The disposal is written with the lots it consumed and their basis,
and `UNPRICED` where the proceeds and gain would be, until you answer it.
**A sale the history cannot cover is a gap, not a gain.** If a sale is larger
than every lot the scan found before it, the uncovered quantity is marked
`UNMATCHED`, earns no gain, and holds the report open. Selling what the scanned
history never bought is not a sale at zero cost.
**Corporate actions are not disposals.** A split runs as burn-and-mint here,
which to a log reader looks like selling everything and rebuying at no cost.
Quantity and cost-per-unit are adjusted; total cost basis and the acquisition
date are preserved. It is applied once, to the lots still open when it lands; a
lot bought afterwards is already in the new units.
**Every method keeps its own lots.** The history is walked once, in chain order.
A sale is matched only against lots bought before it, and FIFO, LIFO, HIFO and
average each sell from their own copy of the inventory, so a second LIFO sale
sees what the first LIFO sale left. The open-lots sheet follows the method you
export under.
**Gas is yours when you sent the transaction.** Each receipt states
`gasUsed × effectiveGasPrice`. On a buy or a sale one of your wallets sent, that
joins the basis of what it bought or comes off the proceeds of what it sold. A
move between your wallets records its gas on the lots it carried. A transfer
someone sent you, an issuer's split or distribution, and a bundler's transaction
were paid for by their senders and add nothing. Gas is valued at the ETH rate,
so with `--eth=off`, or in a window with no rate, it counts as zero.
---
## What it will not do
- Hold your keys, sign anything, or move anything. It only reads.
- Guess a price it cannot derive.
- Book a transfer between your own wallets as a sale.
- Read a split as a disposal.
- Quietly drop a transaction it could not understand.
- Tell you what you owe. It computes figures; rates and eligibility are yours
to confirm against current regulation where you live.
**This is not tax advice, and it is not a substitute for an accountant.**
---
## Known limits
These are real and currently unsolved. They are listed because a tool that
hides its edges is worse than one that names them.
- **Corporate actions are untested against a real event.** The detection is
written and covered by tests, but no split has been observed on this chain
yet, so the code path has never met a live one.
- **ETH moved by a contract mid-transaction is not read.** It leaves no
Transfer log and no `tx.value`, and the public RPC offers no `debug_` or
`trace_`. That node can re-simulate a transaction with `eth_simulateV1` for
about ten minutes, which would recover it; Vrismcost does not do this yet, and
after those ten minutes the public RPC cannot recover it at all.
- **Rebases leave nothing to detect.** A balance that changes with no Transfer
log cannot be seen. Splits and migrations are caught by their burn-and-mint
pattern; a silent rebase is not.
- **Very active addresses hit a wall.** `eth_getLogs` refuses more than 10,000
matching logs. Below a 500-block window there is nowhere left to narrow to,
and the scan stops and says so rather than truncating.
- **A full-history scan takes a couple of minutes** — 134 seconds for one
active wallet when last measured. The slowest stage is reading ETH rates off
the pools.
- **One value across several assets is split evenly.** When an answered
transaction acquired more than one token, nothing on chain states their
relative worth. The split is recorded as an assumption on the lot.
- **A smart account's gas is not read.** Its operations are sent by a bundler,
which recovers the fee from the account inside the same transaction. Gas
counts only on transactions your wallets sent, so a smart account's basis
carries none of it.
- **An ETH leg is priced from its window's centre.** The rate is derived once
per 100,000 blocks, at the middle of that range, so the rate applied to a
trade can come from up to about 1.4 hours away. On a day ETH moves 2%, that
gap is visible in the basis. Smaller windows cost more queries against a
rate-limited node; this is the current trade.
- **No tax-year bucketing.** Robinhood Chain reached production on 30 June
2026, so 2026 is its only tax year and "since the chain went live" is the
same figure. This becomes necessary in January 2027.
---
## Releases
**A published version never changes.** If the code needs fixing, the version
number moves with it. The site's build refuses to publish different bytes under
a version that is already out, because from the outside that is
indistinguishable from tampering, whatever the reason for it. Every release is
appended to `RELEASES` on the site and never edited, and every file in this
archive is listed with its hash in `SHA256SUMS` there.
---
## Layout
vrismcost.py the command line, and run(): the whole pipeline in one function
fetcher.py reads the chain — logs, timestamps, tx.value, gas
engine.py classification, lots, matching, corporate actions
price.py derives the ETH rate from WETH/USDG pools
report.py the CSV export
vault.py the encrypted local store
test_vrismcost.py the suite
README.md this file
requirements.txt the one dependency
Run the suite after any change:
```bash
python test_vrismcost.py -v
```
Every test runs offline, in a few seconds. They exist because a tax tool does
not crash when it is wrong — it returns a number, confidently, and the number
is wrong.
---
## Measurements
Everything above is derived from the chain as measured on **12 September 2026**.
Chain facts change; re-measure before relying on them.
| | |
| --- | --- |
| Production since | 30 June 2026 (block 505,859) |
| Block time | 0.100s steady |
| State retention | ~10 minutes — not an archive node |
| Log retention | Readable where checked, 30,000,000 blocks back |
| `eth_getLogs` cap | 10,000 matched logs per query |
| Legs per transaction | 5.20 mean, 70.9% above two (884 transactions over 200 blocks) |
| Transactions moving ETH | 20% (same sample) |
| Indexed API | None — Blockscout returns 403 |
| Browser access | `access-control-allow-origin: *` |
Block 1 carries a stamp of 30 April 2026, but the chain crawled at up to 195s
per block until it reached production cadence. That stamp is a bring-up
artefact, not a launch date, and using it would overstate every holding period
by two months.
# Vrismcost — one dependency.
#
# AES-GCM for the encrypted vault. It has to come from an audited
# implementation; composing an AEAD out of hashlib would be the kind of
# cleverness that loses somebody their tax records.
#
# Everything else — the chain reader, the engine, the CSV export — is
# standard library only.
cryptography>=42.0