""" 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, }