Initial scaffold + canonical Activity model
Context ------- New connector suite that syncs UK brokerage activity (Trading212, InvestEngine, Schwab email-parsed, CSV drop-folder) into Wealthfolio. Lives outside finance/ intentionally — finance/ is untouched per the plan at ~/.claude/plans/let-s-work-on-linking-temporal-valiant.md. This change ----------- - Poetry project with httpx, typer, bs4, dev tools (pytest, mypy strict, ruff, yapf). - Canonical Activity + Account models with the 6 UK tax wrappers (ISA/SIPP/GIA/LISA/JISA/WORKPLACE_PENSION) and the 12 Wealthfolio activity types from docs/activities/activity-types.md on the upstream. - Validation invariants: BUY/SELL need qty+price, DIVIDEND/DEPOSIT/etc need amount — raises early so providers can't silently emit broken rows. - to_wealthfolio_csv_row() shape matches Wealthfolio's CSV import; primary sink path per the plan. Test plan --------- ## Automated - poetry run pytest -q → 7 passed in 0.03s - poetry run mypy broker_sync tests → Success: no issues found in 4 source files - poetry run ruff check . → All checks passed! - poetry run yapf --diff --recursive broker_sync tests → no diff ## Manual Verification Not applicable — pure data model, no runtime behaviour. Closes: code-thw.1
This commit is contained in:
commit
a2aa7ec486
7 changed files with 958 additions and 0 deletions
105
broker_sync/models.py
Normal file
105
broker_sync/models.py
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
from enum import StrEnum
|
||||
|
||||
|
||||
class AccountType(StrEnum):
|
||||
ISA = "ISA"
|
||||
SIPP = "SIPP"
|
||||
GIA = "GIA"
|
||||
LISA = "LISA"
|
||||
JISA = "JISA"
|
||||
WORKPLACE_PENSION = "WORKPLACE_PENSION"
|
||||
|
||||
|
||||
class ActivityType(StrEnum):
|
||||
BUY = "BUY"
|
||||
SELL = "SELL"
|
||||
DIVIDEND = "DIVIDEND"
|
||||
INTEREST = "INTEREST"
|
||||
DEPOSIT = "DEPOSIT"
|
||||
WITHDRAWAL = "WITHDRAWAL"
|
||||
TRANSFER_IN = "TRANSFER_IN"
|
||||
TRANSFER_OUT = "TRANSFER_OUT"
|
||||
CONVERSION_IN = "CONVERSION_IN"
|
||||
CONVERSION_OUT = "CONVERSION_OUT"
|
||||
FEE = "FEE"
|
||||
TAX = "TAX"
|
||||
|
||||
|
||||
class FxRateSource(StrEnum):
|
||||
ECB_LIVE = "ECB_LIVE"
|
||||
HMRC_MONTHLY = "HMRC_MONTHLY"
|
||||
|
||||
|
||||
_QTY_PRICE_TYPES = {ActivityType.BUY, ActivityType.SELL}
|
||||
_AMOUNT_TYPES = {
|
||||
ActivityType.DIVIDEND,
|
||||
ActivityType.INTEREST,
|
||||
ActivityType.DEPOSIT,
|
||||
ActivityType.WITHDRAWAL,
|
||||
ActivityType.FEE,
|
||||
ActivityType.TAX,
|
||||
ActivityType.TRANSFER_IN,
|
||||
ActivityType.TRANSFER_OUT,
|
||||
ActivityType.CONVERSION_IN,
|
||||
ActivityType.CONVERSION_OUT,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Account:
|
||||
id: str
|
||||
name: str
|
||||
account_type: AccountType
|
||||
currency: str
|
||||
provider: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class Activity:
|
||||
external_id: str
|
||||
account_id: str
|
||||
account_type: AccountType
|
||||
date: datetime
|
||||
activity_type: ActivityType
|
||||
currency: str
|
||||
symbol: str | None = None
|
||||
quantity: Decimal | None = None
|
||||
unit_price: Decimal | None = None
|
||||
amount: Decimal | None = None
|
||||
fee: Decimal = field(default_factory=lambda: Decimal("0"))
|
||||
amount_gbp: Decimal | None = None
|
||||
fx_rate_gbp: Decimal | None = None
|
||||
fx_rate_source: FxRateSource | None = None
|
||||
notes: str | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.activity_type in _QTY_PRICE_TYPES and (
|
||||
self.quantity is None or self.unit_price is None
|
||||
):
|
||||
raise ValueError(f"{self.activity_type} requires quantity and unit_price")
|
||||
if self.activity_type in _AMOUNT_TYPES and self.amount is None:
|
||||
raise ValueError(f"{self.activity_type} requires amount")
|
||||
|
||||
def to_wealthfolio_csv_row(self) -> dict[str, str]:
|
||||
return {
|
||||
"date": self.date.isoformat(),
|
||||
"activity_type": str(self.activity_type),
|
||||
"symbol": self.symbol or "",
|
||||
"quantity": _fmt(self.quantity),
|
||||
"unit_price": _fmt(self.unit_price),
|
||||
"currency": self.currency,
|
||||
"fee": _fmt(self.fee),
|
||||
"amount": _fmt(self.amount),
|
||||
"notes": self.notes or "",
|
||||
}
|
||||
|
||||
|
||||
def _fmt(v: Decimal | None) -> str:
|
||||
if v is None:
|
||||
return ""
|
||||
return format(v, "f")
|
||||
Loading…
Add table
Add a link
Reference in a new issue