""" 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 own 2568.72 2026-07-01 answer the queue python vrismcost.py vault answer income 84.60 python vrismcost.py vault answer exclude python vrismcost.py vault forget 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]}… [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))