feat(ledger): связь перевода ZenMoney с брокерским пополнением
Случай C из плана §1.6: без него одни и те же деньги считаются дважды, а перевод на брокерский счёт выглядит расходом. match_flows — чистая функция над двумя последовательностями кандидатов, как apply() в lots.py; rebuild_flow_links её обвязка с сессией. Скоринг: та же валюта, тот же счёт, |Δ| ≤ max(1 ₽, 0,5 %), разрыв ≤ 3 РАБОЧИХ дня. Рабочих, а не календарных: деньги на брокерский счёт в субботу не приходят, и календарное окно систематически теряло бы пятничные переводы. Праздники не моделируются — лишний праздник делает матчер строже, а не наглее. Направление выбрано по данным, а не по тексту плана. План говорит «outcome на зеркальный счёт = пополнение», но зеркальный ZM-счёт это отражение брокерского кэша: деньги идут «карта → зеркало» (income на зеркале), а на брокере в тот же день deposit. На живых данных income-соглашение даёт 525 пар, обратное — 4. Для маршрута через правило broker_target, где брокера в ZenMoney нет вовсе, направление остаётся как в плане: outcome ↔ deposit. Жадность 1:1 по (дельта суммы, разрыв в днях): пара берётся, только если свободны обе стороны. Ручные линки не пересобираются — обе их стороны исключаются из пулов, иначе автоматика молча переписывала бы решение человека. Связанная транзакция получает flow_type = internal_transfer здесь, а не в classify: классификатор не может знать о линках, которых на момент его работы ещё нет.
This commit is contained in:
Dmitry
@@ -0,0 +1,526 @@
|
||||
"""Pair a ZenMoney transfer with the broker deposit/withdrawal it actually was (plan §1.6 C).
|
||||
|
||||
The same money arrives twice: ZenMoney sees a transfer leaving a card, the broker sees a
|
||||
`deposit` on its own ledger. Nothing joins them, so net worth counts both, and a top-up of a
|
||||
brokerage account reads as an expense — the asymmetry "I invested, therefore I got poorer"
|
||||
that fin-dashboard could only report. A `flow_link` row is the join, and once it exists the
|
||||
ZenMoney side is reclassified as `internal_transfer`.
|
||||
|
||||
There is no shared identifier to join on — ZenMoney does not know the broker's operation id —
|
||||
so the pairing is a scored guess over three facts, in the order they matter:
|
||||
|
||||
* **the broker account is known**, from `account_link`, from `account.mirror_of_account_id`
|
||||
or from a `broker_target` rule. A pair is never considered across accounts;
|
||||
* **the amount**, within `max(1 ₽, 0.5 %)` — a transfer can lose a rounding kopeck, and a
|
||||
currency-conversion top-up arrives a little off;
|
||||
* **the date, in BUSINESS days**. Money sent on Friday lands on Monday: counting calendar
|
||||
days would either reject that pair or, with a wider window, start matching a different
|
||||
week's top-up of the same round sum.
|
||||
|
||||
Matching is greedy and strictly 1:1 (both unique constraints on `flow_link` enforce it):
|
||||
pairs are ranked best-first and taken while both sides are still free. Two candidates for the
|
||||
same 10 000 ₽ therefore produce one link and one leftover, never two half-links — and the
|
||||
leftover is reported, not guessed at. Everything that stays unmatched is a FINDING and, later,
|
||||
`GET /links/unmatched` for manual pairing.
|
||||
|
||||
Direction, read off the live data rather than assumed: a ZenMoney account that mirrors a
|
||||
broker account holds the broker's cash, so money moving INTO it is a broker `deposit` and
|
||||
money leaving it is a `withdrawal`. The `broker_target` rule covers the opposite shape — a
|
||||
brokerage that ZenMoney does not model at all — where the transfer is a plain outflow from an
|
||||
ordinary card, and there an outcome is the deposit.
|
||||
|
||||
`match_flows` is pure: two sequences of light dataclasses in, pairs out, no session. All the
|
||||
rules above are tested on synthetic candidates; `rebuild_flow_links` below is the thin I/O
|
||||
wrapper that loads candidates, runs it, and replaces the automatic links.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import enum
|
||||
import logging
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import date, timedelta
|
||||
from decimal import Decimal
|
||||
|
||||
from sqlalchemy import delete, insert, or_, select, update
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from fintracker.analytics import FINDINGS
|
||||
from fintracker.models import (
|
||||
Account,
|
||||
AccountKind,
|
||||
AccountLink,
|
||||
CashTxn,
|
||||
Event,
|
||||
EventKind,
|
||||
EventStatus,
|
||||
FlowLink,
|
||||
FlowLinkKind,
|
||||
FlowType,
|
||||
Rule,
|
||||
RuleKind,
|
||||
RuleMatchType,
|
||||
)
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
ZERO = Decimal(0)
|
||||
ONE = Decimal(1)
|
||||
|
||||
MIN_TOLERANCE = Decimal("1")
|
||||
"""Absolute amount tolerance: a rounding kopeck must not cost a link."""
|
||||
REL_TOLERANCE = Decimal("0.005")
|
||||
"""Relative amount tolerance (0.5 %), which takes over on large transfers."""
|
||||
MAX_GAP_BUSINESS_DAYS = 3
|
||||
"""Settlement window. Business days only — nothing reaches a broker over a weekend."""
|
||||
|
||||
|
||||
class FlowDirection(enum.StrEnum):
|
||||
"""Which way the money crossed the brokerage boundary."""
|
||||
|
||||
to_broker = "to_broker"
|
||||
"""ZenMoney transfer in, broker `deposit`."""
|
||||
from_broker = "from_broker"
|
||||
"""Broker `withdrawal`, ZenMoney money coming back."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ZmCandidate:
|
||||
"""What the matcher needs from a `cash_txn` — deliberately not the ORM object."""
|
||||
|
||||
cash_txn_id: int
|
||||
date: date
|
||||
amount: Decimal
|
||||
"""Always positive: the magnitude that crossed the boundary."""
|
||||
currency: str
|
||||
direction: FlowDirection
|
||||
broker_account_id: int
|
||||
"""Which broker account this transfer is about; resolved by `route`."""
|
||||
route: str
|
||||
"""account_link | mirror | rule — how the broker account was identified."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BrokerCandidate:
|
||||
"""A `deposit` / `withdrawal` event with no link yet."""
|
||||
|
||||
event_id: int
|
||||
account_id: int
|
||||
date: date
|
||||
amount: Decimal
|
||||
"""Always positive; the sign lives in `direction`."""
|
||||
currency: str
|
||||
direction: FlowDirection
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FlowMatch:
|
||||
zm: ZmCandidate
|
||||
broker: BrokerCandidate
|
||||
amount_delta: Decimal
|
||||
day_gap: int
|
||||
"""Business days between the two dates."""
|
||||
confidence: Decimal
|
||||
"""0..1; 1 is the same day and the same kopeck."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class MatchResult:
|
||||
matches: list[FlowMatch] = field(default_factory=list)
|
||||
unmatched_zm: list[ZmCandidate] = field(default_factory=list)
|
||||
unmatched_broker: list[BrokerCandidate] = field(default_factory=list)
|
||||
|
||||
|
||||
def business_days_between(a: date, b: date) -> int:
|
||||
"""Weekdays strictly after the earlier date, up to and including the later one.
|
||||
|
||||
Friday -> Monday is 1, not 3: that is the whole point of counting this way. Bank holidays
|
||||
are not modelled — the window is small and an extra holiday only makes the matcher
|
||||
slightly stricter, which is the safe direction for a guess.
|
||||
"""
|
||||
lo, hi = (a, b) if a <= b else (b, a)
|
||||
days = 0
|
||||
current = lo
|
||||
while current < hi:
|
||||
current += timedelta(days=1)
|
||||
if current.weekday() < 5:
|
||||
days += 1
|
||||
return days
|
||||
|
||||
|
||||
def amount_tolerance(amount: Decimal) -> Decimal:
|
||||
"""How far two amounts may differ and still be the same money."""
|
||||
return max(MIN_TOLERANCE, abs(amount) * REL_TOLERANCE)
|
||||
|
||||
|
||||
def score(
|
||||
zm: ZmCandidate, broker: BrokerCandidate, *, max_gap: int = MAX_GAP_BUSINESS_DAYS
|
||||
) -> FlowMatch | None:
|
||||
"""A pair, or None when the two cannot be the same money.
|
||||
|
||||
Confidence is only a ranking and review aid — the hard filters below are what decide.
|
||||
"""
|
||||
if zm.broker_account_id != broker.account_id:
|
||||
return None
|
||||
if zm.direction is not broker.direction:
|
||||
return None
|
||||
if zm.currency != broker.currency:
|
||||
return None
|
||||
|
||||
delta = abs(zm.amount - broker.amount)
|
||||
tolerance = amount_tolerance(max(zm.amount, broker.amount))
|
||||
if delta > tolerance:
|
||||
return None
|
||||
|
||||
gap = business_days_between(zm.date, broker.date)
|
||||
if gap > max_gap:
|
||||
return None
|
||||
|
||||
# a coefficient, so the arithmetic may be sloppy — but it is stored, hence Decimal
|
||||
confidence = (
|
||||
ONE
|
||||
- Decimal("0.4") * (delta / tolerance)
|
||||
- Decimal("0.3") * (Decimal(gap) / Decimal(max_gap) if max_gap else ZERO)
|
||||
)
|
||||
return FlowMatch(
|
||||
zm=zm,
|
||||
broker=broker,
|
||||
amount_delta=delta,
|
||||
day_gap=gap,
|
||||
confidence=confidence.quantize(Decimal("0.0001")),
|
||||
)
|
||||
|
||||
|
||||
def match_flows(
|
||||
zm_candidates: Sequence[ZmCandidate],
|
||||
broker_candidates: Sequence[BrokerCandidate],
|
||||
*,
|
||||
max_gap: int = MAX_GAP_BUSINESS_DAYS,
|
||||
) -> MatchResult:
|
||||
"""Greedy 1:1 pairing, best pairs first: smaller amount delta, then smaller day gap.
|
||||
|
||||
Ordering by delta before gap is deliberate. An exact sum three days later is the same
|
||||
money far more often than an approximate sum on the same day — people transfer round
|
||||
numbers, and two round numbers on one day are genuinely ambiguous.
|
||||
"""
|
||||
pairs = [
|
||||
match
|
||||
for zm in zm_candidates
|
||||
for broker in broker_candidates
|
||||
if (match := score(zm, broker, max_gap=max_gap)) is not None
|
||||
]
|
||||
pairs.sort(key=lambda m: (m.amount_delta, m.day_gap, m.zm.cash_txn_id, m.broker.event_id))
|
||||
|
||||
result = MatchResult()
|
||||
used_zm: set[int] = set()
|
||||
used_broker: set[int] = set()
|
||||
for pair in pairs:
|
||||
if pair.zm.cash_txn_id in used_zm or pair.broker.event_id in used_broker:
|
||||
continue
|
||||
used_zm.add(pair.zm.cash_txn_id)
|
||||
used_broker.add(pair.broker.event_id)
|
||||
result.matches.append(pair)
|
||||
|
||||
result.unmatched_zm = [c for c in zm_candidates if c.cash_txn_id not in used_zm]
|
||||
result.unmatched_broker = [c for c in broker_candidates if c.event_id not in used_broker]
|
||||
return result
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- I/O
|
||||
|
||||
|
||||
async def rebuild_flow_links(session: AsyncSession) -> None:
|
||||
"""Replace the automatic links and reclassify the transfers they explain.
|
||||
|
||||
Manual links are the user's judgement and are never rebuilt: both their sides are taken
|
||||
out of the candidate pools before matching, so the matcher can only ever fill the gaps
|
||||
around them.
|
||||
"""
|
||||
manual_txns, manual_events = await _manual_links(session)
|
||||
zm_candidates = [
|
||||
c for c in await _load_zm_candidates(session) if c.cash_txn_id not in manual_txns
|
||||
]
|
||||
broker_candidates = [
|
||||
c for c in await _load_broker_candidates(session) if c.event_id not in manual_events
|
||||
]
|
||||
|
||||
result = match_flows(zm_candidates, broker_candidates)
|
||||
|
||||
await session.execute(delete(FlowLink).where(FlowLink.kind == FlowLinkKind.auto))
|
||||
if result.matches:
|
||||
await session.execute(
|
||||
insert(FlowLink),
|
||||
[
|
||||
{
|
||||
"cash_txn_id": m.zm.cash_txn_id,
|
||||
"event_id": m.broker.event_id,
|
||||
"kind": FlowLinkKind.auto,
|
||||
"confidence": m.confidence,
|
||||
"amount_delta": m.amount_delta,
|
||||
"currency": m.zm.currency,
|
||||
"day_gap": m.day_gap,
|
||||
"note": f"{m.zm.route}/{m.zm.direction.value}",
|
||||
}
|
||||
for m in result.matches
|
||||
],
|
||||
)
|
||||
|
||||
# the point of the whole module: a linked transfer is not spending. `classify` has
|
||||
# already rebuilt every flow_type from the rules, so this overwrite is the last word
|
||||
# for exactly the transactions a link explains — including the manual ones.
|
||||
linked_txns = {m.zm.cash_txn_id for m in result.matches} | manual_txns
|
||||
if linked_txns:
|
||||
await session.execute(
|
||||
update(CashTxn)
|
||||
.where(CashTxn.id.in_(linked_txns), CashTxn.deleted.is_(False))
|
||||
.values(flow_type=FlowType.internal_transfer)
|
||||
.execution_options(synchronize_session=False)
|
||||
)
|
||||
|
||||
_report(result)
|
||||
log.info(
|
||||
"flow links: %s matched, %s ZenMoney and %s broker candidates left over",
|
||||
len(result.matches),
|
||||
len(result.unmatched_zm),
|
||||
len(result.unmatched_broker),
|
||||
)
|
||||
|
||||
|
||||
async def _manual_links(session: AsyncSession) -> tuple[set[int], set[int]]:
|
||||
rows = (
|
||||
await session.execute(
|
||||
select(FlowLink.cash_txn_id, FlowLink.event_id).where(
|
||||
FlowLink.kind == FlowLinkKind.manual
|
||||
)
|
||||
)
|
||||
).all()
|
||||
return {r.cash_txn_id for r in rows}, {r.event_id for r in rows}
|
||||
|
||||
|
||||
async def _broker_targets(session: AsyncSession) -> dict[int, tuple[int, str]]:
|
||||
"""ZenMoney account -> (broker account, route), explicit mapping winning over the mirror."""
|
||||
targets: dict[int, tuple[int, str]] = {}
|
||||
mirrors = (
|
||||
await session.execute(
|
||||
select(Account.id, Account.mirror_of_account_id).where(
|
||||
Account.mirror_of_account_id.is_not(None)
|
||||
)
|
||||
)
|
||||
).all()
|
||||
for row in mirrors:
|
||||
if row.mirror_of_account_id is not None:
|
||||
targets[row.id] = (row.mirror_of_account_id, "mirror")
|
||||
|
||||
links = (await session.execute(select(AccountLink))).scalars().all()
|
||||
for link in links:
|
||||
targets[link.zm_account_id] = (link.broker_account_id, "account_link")
|
||||
return targets
|
||||
|
||||
|
||||
async def _rule_targets(session: AsyncSession) -> list[tuple[Rule, int]]:
|
||||
"""Enabled `broker_target` rules whose value names a real broker account."""
|
||||
rules = (
|
||||
(
|
||||
await session.execute(
|
||||
select(Rule)
|
||||
.where(Rule.kind == RuleKind.broker_target, Rule.enabled.is_(True))
|
||||
.order_by(Rule.priority, Rule.id)
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
if not rules:
|
||||
return []
|
||||
|
||||
broker_ids = set(
|
||||
(await session.execute(select(Account.id).where(Account.kind == AccountKind.broker)))
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
resolved: list[tuple[Rule, int]] = []
|
||||
for rule in rules:
|
||||
try:
|
||||
account_id = int(rule.value or "")
|
||||
except ValueError:
|
||||
account_id = 0
|
||||
if account_id in broker_ids:
|
||||
resolved.append((rule, account_id))
|
||||
else:
|
||||
FINDINGS.add(
|
||||
"flow_link_bad_rule",
|
||||
"warn",
|
||||
f"Правило broker_target #{rule.id} указывает на несуществующий "
|
||||
f"брокерский счёт {rule.value!r}",
|
||||
ref={"rule_id": rule.id, "value": rule.value},
|
||||
)
|
||||
return resolved
|
||||
|
||||
|
||||
async def _load_zm_candidates(session: AsyncSession) -> list[ZmCandidate]:
|
||||
"""Every transfer that could be the ZenMoney half of a broker cash flow.
|
||||
|
||||
Two shapes, and a transaction may legitimately produce two candidates — a transfer from
|
||||
one brokerage mirror straight into another is a withdrawal and a deposit at once.
|
||||
"""
|
||||
targets = await _broker_targets(session)
|
||||
rules = await _rule_targets(session)
|
||||
if not targets and not rules:
|
||||
return []
|
||||
|
||||
conditions = []
|
||||
if targets:
|
||||
conditions.append(CashTxn.income_account_id.in_(targets))
|
||||
conditions.append(CashTxn.outcome_account_id.in_(targets))
|
||||
query = select(CashTxn).where(CashTxn.deleted.is_(False))
|
||||
if not rules:
|
||||
# without rules only the mapped accounts can produce a candidate, so let Postgres
|
||||
# do the filtering; a rule may match on payee or comment and has to be evaluated row
|
||||
# by row, and the personal volume (10^4 rows) makes that cheap enough
|
||||
query = query.where(or_(*conditions))
|
||||
txns = (await session.execute(query)).scalars().all()
|
||||
|
||||
candidates: list[ZmCandidate] = []
|
||||
for txn in txns:
|
||||
mapped = False
|
||||
income_target = targets.get(txn.income_account_id or 0)
|
||||
if income_target and txn.income > ZERO:
|
||||
# money landed on the ZenMoney twin of a broker account: that is a deposit
|
||||
candidates.append(
|
||||
ZmCandidate(
|
||||
cash_txn_id=txn.id,
|
||||
date=txn.date,
|
||||
amount=txn.income,
|
||||
currency=txn.income_currency or "RUB",
|
||||
direction=FlowDirection.to_broker,
|
||||
broker_account_id=income_target[0],
|
||||
route=income_target[1],
|
||||
)
|
||||
)
|
||||
mapped = True
|
||||
outcome_target = targets.get(txn.outcome_account_id or 0)
|
||||
if outcome_target and txn.outcome > ZERO:
|
||||
candidates.append(
|
||||
ZmCandidate(
|
||||
cash_txn_id=txn.id,
|
||||
date=txn.date,
|
||||
amount=txn.outcome,
|
||||
currency=txn.outcome_currency or "RUB",
|
||||
direction=FlowDirection.from_broker,
|
||||
broker_account_id=outcome_target[0],
|
||||
route=outcome_target[1],
|
||||
)
|
||||
)
|
||||
mapped = True
|
||||
if mapped:
|
||||
continue
|
||||
|
||||
rule_target = _first_matching_rule(rules, txn)
|
||||
if rule_target is None:
|
||||
continue
|
||||
# the brokerage is invisible to ZenMoney, so the transfer is a one-sided outflow
|
||||
# (or refund) on an ordinary account and the rule says where it went
|
||||
if txn.outcome > ZERO:
|
||||
candidates.append(
|
||||
ZmCandidate(
|
||||
cash_txn_id=txn.id,
|
||||
date=txn.date,
|
||||
amount=txn.outcome,
|
||||
currency=txn.outcome_currency or "RUB",
|
||||
direction=FlowDirection.to_broker,
|
||||
broker_account_id=rule_target,
|
||||
route="rule",
|
||||
)
|
||||
)
|
||||
elif txn.income > ZERO:
|
||||
candidates.append(
|
||||
ZmCandidate(
|
||||
cash_txn_id=txn.id,
|
||||
date=txn.date,
|
||||
amount=txn.income,
|
||||
currency=txn.income_currency or "RUB",
|
||||
direction=FlowDirection.from_broker,
|
||||
broker_account_id=rule_target,
|
||||
route="rule",
|
||||
)
|
||||
)
|
||||
return candidates
|
||||
|
||||
|
||||
def _first_matching_rule(rules: list[tuple[Rule, int]], txn: CashTxn) -> int | None:
|
||||
"""The broker account of the first `broker_target` rule this transaction matches."""
|
||||
from fintracker.analytics.classify import like_match
|
||||
|
||||
for rule, account_id in rules:
|
||||
match rule.match_type:
|
||||
case RuleMatchType.id:
|
||||
matched = txn.source_id == rule.pattern
|
||||
case RuleMatchType.payee:
|
||||
matched = like_match(rule.pattern, txn.payee)
|
||||
case RuleMatchType.comment:
|
||||
matched = like_match(rule.pattern, txn.comment)
|
||||
case RuleMatchType.account:
|
||||
matched = rule.pattern.isdigit() and int(rule.pattern) in (
|
||||
txn.outcome_account_id,
|
||||
txn.income_account_id,
|
||||
)
|
||||
case _:
|
||||
# category and mcc rules need the tag tree `classify` builds; a broker
|
||||
# top-up is identified by payee, comment or account in practice
|
||||
matched = False
|
||||
if matched:
|
||||
return account_id
|
||||
return None
|
||||
|
||||
|
||||
async def _load_broker_candidates(session: AsyncSession) -> list[BrokerCandidate]:
|
||||
"""Confirmed deposits and withdrawals — the broker's own view of the same money."""
|
||||
rows = (
|
||||
await session.execute(
|
||||
select(Event.id, Event.account_id, Event.trade_date, Event.amount, Event.currency)
|
||||
.where(
|
||||
Event.status == EventStatus.confirmed,
|
||||
Event.kind.in_((EventKind.deposit, EventKind.withdrawal)),
|
||||
)
|
||||
.order_by(Event.trade_date, Event.id)
|
||||
)
|
||||
).all()
|
||||
return [
|
||||
BrokerCandidate(
|
||||
event_id=r.id,
|
||||
account_id=r.account_id,
|
||||
date=r.trade_date,
|
||||
amount=abs(r.amount),
|
||||
currency=r.currency,
|
||||
# the kind and the sign agree in every mapper, but `amount` is the cash effect
|
||||
# and therefore the one that cannot be wrong
|
||||
direction=(FlowDirection.to_broker if r.amount > ZERO else FlowDirection.from_broker),
|
||||
)
|
||||
for r in rows
|
||||
if r.amount != ZERO
|
||||
]
|
||||
|
||||
|
||||
def _report(result: MatchResult) -> None:
|
||||
"""Leftovers are the interesting part: each one is money counted twice somewhere."""
|
||||
if result.unmatched_zm:
|
||||
total = sum((c.amount for c in result.unmatched_zm), start=ZERO)
|
||||
FINDINGS.add(
|
||||
"flow_link_unmatched_zm",
|
||||
"warn",
|
||||
f"{len(result.unmatched_zm)} переводов ZenMoney на брокерские счета "
|
||||
f"({total:.2f}) не нашли парного пополнения/вывода",
|
||||
count=len(result.unmatched_zm),
|
||||
ref={"cash_txn_ids": sorted(c.cash_txn_id for c in result.unmatched_zm)[:50]},
|
||||
)
|
||||
if result.unmatched_broker:
|
||||
total = sum((c.amount for c in result.unmatched_broker), start=ZERO)
|
||||
FINDINGS.add(
|
||||
"flow_link_unmatched_broker",
|
||||
"warn",
|
||||
f"{len(result.unmatched_broker)} пополнений/выводов брокера ({total:.2f}) "
|
||||
"не нашли парного перевода в ZenMoney",
|
||||
count=len(result.unmatched_broker),
|
||||
ref={"event_ids": sorted(c.event_id for c in result.unmatched_broker)[:50]},
|
||||
)
|
||||
Reference in New Issue
Block a user