app/brokercsv.py
Rama main · réplica de lectura del repositorio original.
"""Broker CSV import: parse SelfWealth/CommSec exports into candidate trades.
Neither broker offers an API, so the flow is: download the CSV → upload it on
/imports-exports → review the parsed preview
(duplicates and unknown tickers are flagged) → commit. Column layouts are
format files: shipped in app/formats/brokers, or written by the in-app
mapper (Map a new broker CSV), so a changed export format is mapped again.
Parsing is strict on purpose: an unrecognised header or row fails loudly with
the offending text rather than guessing — first runs against a new broker
export are expected to need a mapping tweak.
"""
from __future__ import annotations
import csv
import datetime as dt
import io
import re
from dataclasses import dataclass, field
from decimal import Decimal
from itertools import accumulate
from sqlalchemy import select
from sqlalchemy.orm import Session
from . import money, queries
from .markettime import to_market
from .models import (
CorporateAction,
Instrument,
Trade,
exchange_problem,
ticker_problem,
trade_order,
)
from .pricefeed import MARKETS
from .settings import BrokerFormat, ImportSettings
from .tenancy import owned
# CommSec transactions export: Date,Reference,Details,Debit($),Credit($),Balance($)
# with trade rows like "B 50 ALPHA @ 94.500000" / "S 174 ACME @ 8.550000".
COMMSEC_DETAILS = re.compile(
r"^\s*(?P<action>[BS])\s+(?P<units>[\d,]+)\s+(?P<ticker>[A-Z0-9]{2,6})\s+@\s+(?P<price>[\d,.]+)\s*$"
)
# Broker exports very commonly pad a date with a zero time —
# "2024-02-13 00:00:00" — so a date-only format has to tolerate one.
# The suffixes are tried in order after the bare format, longest first so
# "%H:%M:%S" is not matched as "%H:%M" with seconds left over.
TIME_SUFFIXES = (" %H:%M:%S", " %H:%M", "T%H:%M:%S", "T%H:%M")
# A strptime pattern as a person reads it, for saying which a date missed.
_WRITTEN = {"%d": "DD", "%m": "MM", "%Y": "YYYY", "%y": "YY", "%b": "Mon", "%B": "Month",
"%H": "hh", "%I": "hh", "%M": "mm", "%S": "ss", "%p": "AM/PM"}
def written_as(date_format: str) -> str:
"""`%d/%m/%Y` as `DD/MM/YYYY`."""
return re.sub(r"%[a-zA-Z]", lambda m: _WRITTEN.get(m.group(0), m.group(0)), date_format)
def strptime_any(text: str, date_format: str) -> dt.datetime:
"""`date_format`, or it with a time on the end."""
try:
return dt.datetime.strptime(text, date_format)
except ValueError:
for suffix in TIME_SUFFIXES:
try:
return dt.datetime.strptime(text, date_format + suffix)
except ValueError:
continue
raise
def in_trading_hours(clock: dt.time | None, exchange: str) -> tuple[dt.time | None, bool]:
"""A trade time inside the exchange's session, and whether it had to move.
Before the open becomes the open, after the close becomes the close. A time
outside the session is not a trading time: it is a padded date ("00:00:00",
which is most exports), a settlement stamp, or a clock in another timezone.
Storing it as given would misorder the day. `time` feeds one thing —
`models.trade_order`, which sequences same-day trades for FIFO — so a
midnight stamp sorts an imported trade ahead of every hand-entered one on
that date, and picks a different parcel for a same-day sale.
The open comes back as None, which means exactly the same thing to
`trade_order` and keeps a column of "10:00" out of the interface.
Markets that never close (crypto) have no session to be outside of, so
their times are taken as given.
"""
if clock is None:
return None, False
market = MARKETS.get(exchange.upper())
if market is None or market.open is None or market.close is None:
return clock, False
if clock < market.open:
return None, True
if clock > market.close:
return market.close, True
return (None, False) if clock == market.open else (clock, False)
def parse_when(value: str, date_format: str) -> tuple[dt.date, dt.time | None]:
"""The date, and the time if the cell carried a real one.
Midnight is no time at all: a broker writing "2024-02-13 00:00:00" is
padding a date, not claiming a trade at midnight. It is separated from a
stated out-of-hours time here — see `in_trading_hours` — so that a
date-only export does not report every one of its rows as adjusted.
"""
stamp = strptime_any(value.strip(), date_format) # raises for the caller
clock = stamp.time()
return stamp.date(), None if clock == dt.time(0) else clock
@dataclass
class CandidateTrade:
date: dt.date
ticker: str
type: str # buy / sell
quantity: Decimal
unit_price: Decimal
brokerage: Decimal
currency: str
exchange: str
time: dt.time | None = None # only when the export carried a real one
status: str = "new" # new / duplicate / unknown-instrument / refused
detail: str = ""
def as_dict(self) -> dict:
return {
"date": self.date.isoformat(),
"time": self.time.isoformat() if self.time else None,
"ticker": self.ticker,
"type": self.type,
"quantity": str(self.quantity),
"unit_price": str(self.unit_price),
"brokerage": str(self.brokerage),
"currency": self.currency,
"exchange": self.exchange,
"status": self.status,
"detail": self.detail,
}
@classmethod
def from_dict(cls, d: dict) -> CandidateTrade:
return cls(
date=dt.date.fromisoformat(d["date"]),
time=dt.time.fromisoformat(d["time"]) if d.get("time") else None,
ticker=d["ticker"],
type=d["type"],
quantity=Decimal(d["quantity"]),
unit_price=Decimal(d["unit_price"]),
brokerage=Decimal(d["brokerage"]),
currency=d["currency"],
exchange=d["exchange"],
status=d["status"],
detail=d["detail"],
)
@dataclass
class ParseResult:
candidates: list[CandidateTrade] = field(default_factory=list)
skipped: list[str] = field(default_factory=list) # non-trade rows (cash etc.)
errors: list[str] = field(default_factory=list)
adjusted: int = 0 # times moved into trading hours; see `in_trading_hours`
def _num(raw: str, column, name: str) -> Decimal:
return money.parse(str(raw).replace(",", "").replace("$", ""), column, name)
def _check_trade_figures(units: Decimal, price: Decimal, brokerage: Decimal) -> None:
"""The trade form's rules, so a file cannot stage what the form would refuse.
Zero is a real price (a bonus issue, a demerger allocation). A negative
quantity is refused rather than flipped: whether it means a sell is the
export's to say, and none of the shipped formats do.
"""
if units <= 0:
raise money.FigureError(f"Units: {units} has to be more than zero")
if price < 0:
raise money.FigureError(f"Price: {price} can't be negative")
if brokerage < 0:
raise money.FigureError(f"Brokerage: {brokerage} can't be negative")
def parse_csv(text: str, broker: str, fmt: BrokerFormat) -> ParseResult:
"""Every trade in an export, with an account of what was skipped.
A field past the csv module's size limit (128 KiB: a file that is not a
CSV at all, or one runaway quote) raises inside the reader, and was a
server error rather than a sentence about the file."""
try:
return _parse_csv(text, broker, fmt)
except csv.Error:
return ParseResult(errors=["The file could not be read as a CSV."])
def _parse_csv(text: str, broker: str, fmt: BrokerFormat) -> ParseResult:
# `restval`: a row shorter than the header (an export's one-line footer)
# reads as blanks, and is skipped by line like any other, instead of
# handing None to the first `.strip()`.
reader = csv.DictReader(io.StringIO(text), restval="")
if reader.fieldnames is None:
return ParseResult(errors=["The file is empty."])
# "Date, Code" names its columns with the spaces. The rows are read by
# the names the header check and the broker designer use: without them.
reader.fieldnames = headers = [h.strip() for h in reader.fieldnames]
result = ParseResult()
if fmt.kind == "mapped":
missing = [v for v in fmt.columns.values() if v not in headers]
if missing:
result.errors.append(
f"This file has no {', '.join(missing)} column{'s' if len(missing) > 1 else ''}, "
f"which the {broker} format reads. Pick the broker it came from, or map its "
"columns with Map a new broker CSV.")
return result
col = fmt.columns
for i, row in enumerate(reader, start=2):
try:
action = trade_type(row[col["action"]], fmt.actions)
if action is None:
result.skipped.append(f"line {i}: action {row[col['action']]!r}")
continue
# A DRP allotment carries units and no price: the registry
# bought them out of the distribution, and the cost base is in
# the statement rather than this export. Booking it at zero
# would understate the cost base of every parcel and overstate
# the gain on sale — decisions.md #100.
if action == "drp" and not str(row.get(col["price"], "")).strip():
result.skipped.append(
f"line {i}: DRP allotment with no price — import the "
f"dividend statement, which carries the cost base")
continue
written = row[col["date"]]
try:
when, clock = parse_when(written, fmt.date_format)
except ValueError:
result.errors.append(f"line {i}: {written.strip()!r} is not a date "
f"written as {written_as(fmt.date_format)}")
continue
if clock and fmt.times_zone:
when, clock = to_market(clock, when, fmt.times_zone, fmt.exchange)
clock, moved = in_trading_hours(clock, fmt.exchange)
result.adjusted += moved
units = _num(row[col["units"]], Trade.quantity, "Units")
price = _num(row[col["price"]], Trade.unit_price, "Price")
brokerage = _num(row.get(col.get("brokerage", ""), "0") or "0",
Trade.brokerage, "Brokerage")
_check_trade_figures(units, price, brokerage)
result.candidates.append(
CandidateTrade(
date=when,
time=clock,
ticker=row[col["ticker"]].strip().upper(),
type=action,
quantity=units,
unit_price=price,
brokerage=brokerage,
currency=fmt.currency,
exchange=fmt.exchange,
)
)
except (money.FigureError, KeyError, ValueError) as exc:
result.errors.append(f"line {i}: {exc}")
return result
if fmt.kind == "commsec_transactions":
if "Details" not in headers or "Debit($)" not in headers:
result.errors.append(
"This is not CommSec's transactions export header, which reads Date, "
"Reference, Details, Debit($), Credit($), Balance($). Pick the broker the "
"file came from.")
return result
for i, row in enumerate(reader, start=2):
details = (row.get("Details") or "").strip()
m = COMMSEC_DETAILS.match(details)
if not m:
result.skipped.append(f"line {i}: {details!r}")
continue
try:
qty = _num(m.group("units"), Trade.quantity, "Units")
price = _num(m.group("price"), Trade.unit_price, "Price")
_check_trade_figures(qty, price, Decimal(0))
is_buy = m.group("action") == "B"
# Brokerage isn't a column: recover it from the cash movement.
# buy: debit = qty*price + brokerage; sell: credit = qty*price - brokerage.
gross = qty * price
try:
when, clock = parse_when(row["Date"], fmt.date_format)
except ValueError:
result.errors.append(f"line {i}: {row['Date'].strip()!r} is not a date "
f"written as {written_as(fmt.date_format)}")
continue
clock, moved = in_trading_hours(clock, fmt.exchange)
result.adjusted += moved
cash = (_num(row.get("Debit($)") or "0", Trade.quantity, "Debit")
if is_buy else
_num(row.get("Credit($)") or "0", Trade.quantity, "Credit"))
brokerage = (cash - gross) if is_buy else (gross - cash)
if brokerage < 0 or brokerage > max(Decimal("100"), gross / 10):
result.errors.append(
f"line {i}: implausible brokerage {brokerage} from {details!r} "
f"(cash {cash}, gross {gross}) — check the export"
)
continue
result.candidates.append(
CandidateTrade(
date=when,
time=clock,
ticker=m.group("ticker"),
type="buy" if is_buy else "sell",
quantity=qty,
unit_price=price,
brokerage=brokerage.quantize(Decimal("0.01")),
currency=fmt.currency,
exchange=fmt.exchange,
)
)
except (money.FigureError, ValueError) as exc:
result.errors.append(f"line {i}: {exc}")
return result
result.errors.append(f"unknown broker format kind {fmt.kind!r}")
return result
BUILTIN_ACTIONS = {"buy": "buy", "b": "buy", "sell": "sell", "s": "sell"}
def trade_type(raw: str, mapped: dict[str, str] | None = None) -> str | None:
"""This broker's word for a trade type, as one of ours, or None to skip.
The format's own vocabulary wins, matched without case, so a broker calling
a DRP allotment "In" needs no code. Everything else falls back to the words
every broker agrees on.
"""
word = raw.strip()
for source, target in (mapped or {}).items():
if source.strip().lower() == word.lower():
return target if target in ("buy", "sell", "drp") else None
return BUILTIN_ACTIONS.get(word.lower())
def unresolved(candidates: list[CandidateTrade]) -> list[tuple[str, str, int]]:
"""The distinct (ticker, exchange) pairs that would be created, with counts.
Grouped rather than per row: a file with ninety trades in six instruments
asks six questions, not ninety.
"""
seen: dict[tuple[str, str], int] = {}
for c in candidates:
if c.status == "unknown-instrument":
seen[(c.ticker, c.exchange)] = seen.get((c.ticker, c.exchange), 0) + 1
return [(t, e, n) for (t, e), n in sorted(seen.items())]
def relabel(candidates: list[CandidateTrade], moves: dict[tuple[str, str], tuple[str, str]]) -> int:
"""Apply corrected tickers and exchanges to every row that carried the old.
Returns how many rows moved. A correction is per instrument, so fixing
ALPHA on the wrong exchange fixes every trade of it at once.
"""
changed = 0
for c in candidates:
new = moves.get((c.ticker, c.exchange))
if new and new != (c.ticker, c.exchange):
c.ticker, c.exchange = new
changed += 1
return changed
def _own(session: Session, ticker: str, exchange: str, own_ids: set[int]) -> Instrument | None:
"""This portfolio's instrument for a row, never the catalogue's. Whether
some other portfolio holds a ticker is not this one's to learn (#139), so a
row for one is new here, exactly as for a ticker nobody holds."""
return session.scalars(
select(Instrument).where(Instrument.ticker == ticker, Instrument.exchange == exchange,
Instrument.id.in_(own_ids))
).first()
def annotate(session: Session, candidates: list[CandidateTrade], imports: ImportSettings) -> None:
"""Mark each candidate as new / duplicate / unknown-instrument.
Re-entrant: status and detail are reset first, because the resolve step
calls this again after a correction and a stale "unknown-instrument" would
keep refusing a ticker that now resolves.
A recorded trade answers for one row at most, so a second identical fill
on the day is new. Rows with a time take the trade at that time first; a
trade typed without one sits at market open, so time is never the only
reason a row is new.
"""
matching: list[tuple[CandidateTrade, list[Trade]]] = []
own_ids = queries.portfolio_instrument_ids(session)
for c in candidates:
c.status, c.detail = "new", ""
inst = _own(session, c.ticker, c.exchange, own_ids)
if inst is None:
c.status = "unknown-instrument"
# Refused when new instruments are off: the note under the table
# says what to do, so the row says only what is true of it.
c.detail = "will be created" if imports.allow_new_instruments else "not in this portfolio"
continue
matching.append((c, list(session.scalars(
select(Trade).where(
Trade.instrument_id == inst.id,
Trade.date == c.date,
Trade.type == c.type,
Trade.quantity == c.quantity,
Trade.unit_price == c.unit_price,
).order_by(Trade.id)
))))
claimed: set[int] = set()
for same_time in (True, False):
for c, trades in matching:
if c.status == "duplicate":
continue
dupe = next((t for t in trades if t.id not in claimed
and (not same_time or t.time == c.time)), None)
if dupe is not None:
claimed.add(dupe.id)
c.status = "duplicate"
c.detail = f"matches trade #{dupe.id}"
_refuse_shortfalls(session, candidates)
def _refuse_shortfalls(session: Session, candidates: list[CandidateTrade]) -> None:
"""Refuse a row that would sell more than is held, as the trade form
refuses one typed in. Each holding's rows are walked in date order against
what is already recorded and the rows accepted before them; a refused row
is left out of the walk, so the rest are judged without it.
"""
by_holding: dict[tuple[str, str], list[CandidateTrade]] = {}
for c in candidates:
if c.status in ("new", "unknown-instrument"):
by_holding.setdefault((c.ticker, c.exchange), []).append(c)
own_ids = queries.portfolio_instrument_ids(session)
for (ticker, exchange), rows in by_holding.items():
inst = _own(session, ticker, exchange, own_ids)
held = (list(session.scalars(select(Trade).where(Trade.instrument_id == inst.id)))
if inst is not None else [])
actions = (list(session.scalars(select(CorporateAction)
.where(CorporateAction.instrument_id == inst.id)))
if inst is not None else [])
found = (_shortfalls_walked(held, rows, actions)
if any(a.kind == "split" for a in actions) else _shortfalls(held, rows))
for c, breach in found:
c.status, c.detail = "refused", queries.breach_message(ticker, breach)
def _candidate(c: CandidateTrade) -> Trade:
return Trade(date=c.date, time=c.time, type=c.type, quantity=c.quantity)
def _shortfalls_walked(held: list[Trade], rows: list[CandidateTrade],
actions: list[CorporateAction]) -> list[tuple[CandidateTrade, queries.Breach]]:
"""`_shortfalls`, for a holding with a split: each row walked as the trade
form walks one. The running totals there cannot take a split, which
rounds the holding where it falls (#146); such a holding's rows are few."""
refused, accepted = [], []
for c in sorted(rows, key=lambda c: trade_order(_candidate(c))):
candidate = _candidate(c)
breach = queries.balance_breach(held + accepted, candidate, actions=actions)
if breach is None:
accepted.append(candidate)
else:
refused.append((c, breach))
return refused
def _shortfalls(held: list[Trade], rows: list[CandidateTrade]) -> list[tuple[CandidateTrade, queries.Breach]]:
"""Each row the trade form would refuse, with the breach it would name.
The form walks the whole timeline for its one row; done for each row of a
file, that took minutes over a 40,000-row export. So the timeline is
walked once: a row is checked against the units held at its place, and
against what the recorded trades after it need. Only recorded trades come
after it: the rows accepted so far all sort before it, and the rest are
judged later.
"""
timeline = sorted(
[(t, None) for t in held]
+ [(Trade(date=c.date, time=c.time, type=c.type, quantity=c.quantity), c) for c in rows],
key=lambda pair: trade_order(pair[0]))
change = [t.quantity if t.type in ("buy", "drp") else -t.quantity for t, _ in timeline]
# The recorded trades alone: their running total at each place, and
# `need`, the most those from a place on take from what is held there.
recorded = list(accumulate(Decimal(0) if c else step for (_t, c), step in zip(timeline, change)))
need = [Decimal(0)] * (len(timeline) + 1)
for i in reversed(range(len(timeline))):
need[i] = need[i + 1] if timeline[i][1] else max(Decimal(0), need[i + 1] - change[i])
first_below = _first_below(recorded)
refused, units, broken = [], Decimal(0), None
for i, (t, c) in enumerate(timeline):
after = units + change[i]
if c is None:
# Recorded trades the file never touched can already be short;
# the form names the first, before any row of its own.
if after < 0 and broken is None:
broken = queries.Breach(trade=t, balance=after, held_before=units,
is_candidate=False)
units = after
elif broken is not None:
refused.append((c, broken))
elif after < 0:
refused.append((c, queries.Breach(trade=t, balance=after, held_before=units,
is_candidate=True)))
elif after < need[i + 1]:
k = first_below(i + 1, recorded[i] - after)
left = after + recorded[k] - recorded[i]
refused.append((c, queries.Breach(trade=timeline[k][0], balance=left,
held_before=left - change[k], is_candidate=False)))
else:
units = after
return refused
def _first_below(values: list[Decimal]):
"""`find(start, limit)`: the first index from `start` whose value is below
`limit`, in a handful of steps rather than a scan. `levels[k][i]` is the
least of `values[i : i + 2**k]`, so whole runs that stay at or above the
limit are stepped over, longest first."""
levels = [values]
while 2 ** len(levels) <= len(values):
last, run = levels[-1], 2 ** (len(levels) - 1)
levels.append([min(a, b) for a, b in zip(last, last[run:])])
def find(start: int, limit: Decimal) -> int:
i = start
for k in reversed(range(len(levels))):
if i < len(levels[k]) and levels[k][i] >= limit:
i += 2 ** k
return i
return find
def commit(session: Session, candidates: list[CandidateTrade], imports: ImportSettings, source: str) -> dict[str, int]:
counts = {"inserted": 0, "duplicates_skipped": 0, "instruments_created": 0}
refused = [c for c in candidates if c.status == "refused"]
if refused:
raise ValueError(refused[0].detail)
own_ids = queries.portfolio_instrument_ids(session)
for c in candidates:
if c.status == "duplicate":
counts["duplicates_skipped"] += 1
continue
inst = _own(session, c.ticker, c.exchange, own_ids)
if inst is None:
if not imports.allow_new_instruments:
raise ValueError(f"unknown instrument {c.ticker} ({c.exchange})")
# A CSV is arbitrary input; a malformed ticker must not become a row.
problem = ticker_problem(c.ticker) or exchange_problem(c.exchange)
if problem:
raise ValueError(f"{problem} (got {c.ticker!r} on {c.exchange!r})")
# New to this portfolio, and only now looked for in the shared
# catalogue (#125), which the preview could not tell from new.
inst = session.scalars(
select(Instrument).where(Instrument.ticker == c.ticker,
Instrument.exchange == c.exchange)
).first()
if inst is None:
inst = Instrument(
ticker=c.ticker,
exchange=c.exchange,
currency=c.currency,
asset_class="share",
drp=False,
yahoo_symbol=f"{c.ticker}.AX" if c.exchange == "ASX" else c.ticker,
note=f"created by {source} import",
)
session.add(inst)
session.flush()
counts["instruments_created"] += 1
# A removed one comes back, as it does when a trade is typed in.
inst.active = True
session.add(
owned(
session,
Trade(
instrument=inst,
date=c.date,
# Only when the export carried one; unset means MARKET_OPEN.
**({"time": c.time} if c.time else {}),
type=c.type,
quantity=c.quantity,
unit_price=c.unit_price,
brokerage=c.brokerage,
fx_rate=Decimal(1) if c.currency == "AUD" else None, # feed backfills
note=f"imported from {source} CSV",
),
)
)
counts["inserted"] += 1
session.flush() # the caller's scoped session owns the commit
return counts