feat(engine): 現金流 ingestion 地基 — engine 吃得到現金 + 存 state (refs #171)

skills/fomo-kernel/engine/trade_recap.py

@@ -154,6 +154,81 @@ def load(paths):
     return rows
 
 
+def load_cash_flows(paths):
+    """讀出所有影響現金餘額的 row（含買賣 + deposit/withdrawal/dividend/interest/fee）。
+    load() 只留 BUY/SELL 給行為分析；現金餘額要的是**每一筆現金增減**，靠 Amount 欄
+    （券商流水裡 Amount = 現金帳戶實際變動：買=負、賣/存/息=正、費=負，已含手續費淨額）。
+    #171 PR-1（B 路線帳戶級現金地基）。去重骨架同 load()（跨檔重疊期同一筆只算一次）。
+    回傳 [{date, amount, kind, currency}]，依日期排序。"""
+    KIND = {"BUY": "trade", "SELL": "trade", "REINVEST": "trade",
+            "DEPOSIT": "deposit", "WITHDRAWAL": "withdrawal",
+            "DIVIDEND": "dividend", "INTEREST": "interest", "FEE": "fee"}
+    flows = []
+    seen = set()
+    for p in paths:
+        occ = defaultdict(int)
+        with open(p, newline="", encoding="utf-8-sig") as f:
+            for r in csv.DictReader(f):
+                raw = (r.get("Amount") or "").strip()
+                if not raw:                          # 無 Amount（如純 split/journal）→ 不影響現金
+                    continue
+                try:
+                    amt = float(raw)
+                    d = dt.date.fromisoformat((r.get("TradeDate") or "").strip())
+                except (ValueError, KeyError):
+                    continue
+                if abs(amt) < 1e-9:
+                    continue
+                act = (r.get("Action") or "").strip().upper()
+                rectype = (r.get("RecordType") or "").strip().upper()
+                # kind：先看 Action，退而看 RecordType（有些券商現金流只填 RecordType）
+                kind = KIND.get(act) or (rectype.lower() if rectype in
+                       ("DEPOSIT", "WITHDRAWAL", "DIVIDEND", "INTEREST", "FEE") else "other")
+                cur = (r.get("Currency") or "USD").strip().upper() or "USD"
+                key = (d, round(amt, 2), kind, cur)
+                k = occ[key]; occ[key] += 1          # 同檔同鍵多筆（如同日兩筆股息）各給序號，不互殺
+                rec = key + (k,)
+                if rec in seen:                       # 跨檔重疊 = 同序號重複，跳過
+                    continue
+                seen.add(rec)
+                flows.append(dict(date=d, amount=amt, kind=kind, currency=cur))
+    flows.sort(key=lambda x: x["date"])
+    return flows
+
+
+def cash_position(cash_flows, held_mv, anchor=None, prev_end=None):
+    """帳戶現金地基（#171 PR-1）。
+    - cash_balance：有現金餘額錨點 → 錨點 + 其後現金流（正確，對付 is_complete=False 的不完整 CSV）；
+      無錨點 → 全期 Σamount 並標不可信（假設開戶現金 0，CSV 漏一筆 deposit 就偏，honesty_ledger 揭露）。
+    - cash_weight = 現金 /（持倉市值 + 現金）；分母 ≤0 → None。
+    - recent_net_deposit = 本期（prev_end 後）外部淨流入（deposit − withdrawal），給「這筆錢該不該部署」判讀。
+    幣別：假設 cash_flows.amount 已在聚合幣別（混幣換算由 main 接線層負責）。"""
+    if isinstance(prev_end, str):
+        prev_end = dt.date.fromisoformat(prev_end) if prev_end else None
+    a_date = None
+    if anchor and anchor.get("as_of") and anchor.get("amount") is not None:
+        a_date = anchor["as_of"]
+        if isinstance(a_date, str):
+            a_date = dt.date.fromisoformat(a_date)
+        balance = float(anchor["amount"]) + sum(cf["amount"] for cf in cash_flows if cf["date"] > a_date)
+        source, reliable = "anchored", True
+    else:
+        balance = sum(cf["amount"] for cf in cash_flows)
+        source, reliable = "csv_sum", False
+    denom = held_mv + balance
+    # 無錨點的負現金 = csv_sum 假設破裂（買入為主、入金沒記全），weight 是垃圾 → 不報；denom≤0 亦不報。
+    # 有錨點的負現金 = 真融資（margin debit），weight 負有意義（槓桿曝險），照報。
+    if denom <= 1e-9 or (not reliable and balance < 0):
+        weight = None
+    else:
+        weight = balance / denom
+    recent = sum(cf["amount"] for cf in cash_flows
+                 if cf["kind"] in ("deposit", "withdrawal")
+                 and (prev_end is None or cf["date"] > prev_end))
+    return dict(balance=round(balance, 2), weight=weight, source=source,
+                reliable=reliable, recent_net_deposit=round(recent, 2))
+
+
 def _load_skip_note():
     """#50:把 load() 的靜默丟棄計數組成人話短語(進 meta 行)。全零 → 空字串(不吵)。"""
     s = _LOAD_STATS
@@ -1565,7 +1640,7 @@ def _px(t):
 
 
 def build_state(rows, rts, held, dims, overview, ab, rx, currency_meta=None,
-                avg_down=None, last_px=None, prev_end=None):
+                avg_down=None, last_px=None, prev_end=None, cash=None):
     """把這次復盤收斂成一張薄 JSON 狀態,給「下次對帳上次規矩」用(非給人看的卡)。
     只在 main() 偵測 TR_STATE_OUT 時呼叫並寫出;不設 → 完全不執行,引擎行為零變。
     設計依 requirements §4/§10:
@@ -1615,7 +1690,8 @@ def build_state(rows, rts, held, dims, overview, ab, rx, currency_meta=None,
         mk, mv = next(((k, v) for kw, (k, v) in RULE_METRIC.items() if kw in rule), (None, None))
         commitment = {"rule": rule, "metric_key": mk, "metric_value": mv, "goal": "down"}
     # holdings snapshot（目標3 持倉變化）：per-ticker 絕對值 + position cycle（給 thesis 綁 cycle）。
-    # 只存 shares/cost/avg_cost（確定性）；不存 weight（沒現金+即時價算不準，雙審 gemini#4）。
+    # 只存 shares/cost/avg_cost（確定性）；per-position weight 仍不存（需即時價，跨期不穩）。
+    # 帳戶級 cash_weight 改存頂層 cash 欄位（#171 PR-1：有現金錨點才可信，取代原「沒現金算不準」）。
     cyc = current_cycles(rows)                              # 雙審修：與 positions() 同邏輯（不跌負）+ cycle 序號
     holdings = {t: {"shares": round(sh, 4), "cost": round(c, 2),
                     "avg_cost": round(c / sh, 4) if sh > 1e-9 else None,
@@ -1661,6 +1737,7 @@ def build_state(rows, rts, held, dims, overview, ab, rx, currency_meta=None,
             "is_complete": False,                           # CSV 無法自證完整（雙審 codex#3）：不宣稱完整持倉真相
             "positions": holdings,
         },
+        "cash": cash,                                       # #171 PR-1:帳戶現金地基(balance/weight/source/reliable/recent_net_deposit)。None=未提供;source=csv_sum+reliable=False=無錨點靠 Σamount 近似(honesty 揭露)
         "problem_events": p_events,                         # #137 問題帳:本次規約出的事件(SKILL 收尾 append 進 problems.jsonl)
         "problem_opportunities": p_opps,                    # 各 key 本期有無機會犯(規矩對位的 Opportunity Check)
     }
@@ -1882,6 +1959,21 @@ def main():
             else ("多幣別組合(單一市場)的 α/β 按該市場大盤計" if mixed_ccy else None)),
     }
 
+    # 帳戶現金地基（#171 PR-1）：現金流 + 現金餘額錨點 → cash_position。
+    # held_mv = 持倉市值（聚合幣別，無現價用成本近似，同 dim_diversify）= cash_weight 分母。
+    import json
+    cash_flows = load_cash_flows(paths)
+    if mixed_ccy:                                        # 混幣：現金流各幣別 → USD（對齊聚合視圖）
+        cash_flows = [dict(cf, amount=cf["amount"] * fx.get(cf["currency"], 1.0)) for cf in cash_flows]
+    held_mv = sum((sh * lastpx_u[t]) if lastpx_u.get(t) else c for t, (sh, c) in held_u.items())
+    _ca = os.environ.get("TR_CASH")                     # SKILL Step 0 抓對帳單現金餘額 → JSON {as_of, amount, currency}
+    try:
+        cash_anchor = json.loads(_ca) if _ca else None
+    except (ValueError, TypeError):
+        cash_anchor = None
+    cash_data = cash_position(cash_flows, held_mv, anchor=cash_anchor,
+                              prev_end=os.environ.get("TR_PREV_END") or None)
+
     dm_skip = f"({_DM_SKIPPED} 筆格式錯跳過)" if _DM_SKIPPED else ""
     split_note = f"｜分割調整: {n_adj} 筆" if n_adj else ""
     # JSON 模式(SKILL Step 3 走這條):stdout 純 JSON 給 Claude 寫敘事卡;meta 走 stderr 不污染
@@ -1950,7 +2042,8 @@ def main():
         state = build_state(rows, rts, held, dims, overview, ab, rx,
                             currency_meta=currency_meta,
                             avg_down=avg_down, last_px=last_px,
-                            prev_end=os.environ.get("TR_PREV_END") or None)
+                            prev_end=os.environ.get("TR_PREV_END") or None,
+                            cash=cash_data)
         # TR_PREV_END=上次 review 的 date_end(SKILL 對帳模式傳入)→ behavior 型問題事件
         # 只取其後的新交易(weekly 增量);不設 = 初診全期補齊,問題帳統計冷啟動。
         outdir = os.path.dirname(os.path.abspath(path)) or "."

tests/test_engine_units.py

@@ -764,6 +764,93 @@ def test_fifo_held_full_exit_and_aggregation():
     assert _approx(held_avg["X"][0], sh) and _approx(held_avg["X"][1], cost)  # 全賣後重建 → 兩套一致
 
 
+# ─────────────────── 現金地基(#171 PR-1:B 路線帳戶級)───────────────────
+# load() 只留 BUY/SELL 給行為分析;現金餘額要每一筆現金增減(Amount 欄)。這組鎖 load_cash_flows /
+# cash_position 的核心契約:讀全 Amount 列 + kind 分類、錨點 vs csv_sum、weight 分母、入金過濾。
+
+_CASH_CSV = (
+    "Symbol,Quantity,Price,Action,TradeDate,Amount,RecordType\n"
+    "NVDA,10,100.00,BUY,2024-01-10,-1000.00,Trade\n"
+    "NVDA,5,120.00,SELL,2024-02-10,600.00,Trade\n"
+    ",,,Deposit,2024-01-05,5000.00,Other\n"
+    "KO,,,Dividend,2024-01-20,30.00,Dividend\n"
+    ",,,Interest,2024-01-25,-12.50,Interest\n"
+    ",,,Fee,2024-01-28,-5.00,Fee\n"
+    ",,,Withdrawal,2024-02-15,-2000.00,Other\n"
+)
+
+
+def test_cash_flows_reads_all_amount_rows_with_kind():
+    """load_cash_flows 讀所有 Amount≠0 的列(含買賣+存提息費股利),kind 分類正確——
+    對比 load() 只留 BUY/SELL:現金地基要的是每一筆現金增減。"""
+    p = _write_csv(_CASH_CSV)
+    flows = tr.load_cash_flows([p])
+    os.unlink(p)
+    assert len(flows) == 7, f"7 筆現金流(含買賣),實得 {len(flows)}"
+    kinds = sorted(f["kind"] for f in flows)
+    assert kinds == ["deposit", "dividend", "fee", "interest", "trade", "trade", "withdrawal"], kinds
+    assert _approx(sum(f["amount"] for f in flows), 2612.50), sum(f["amount"] for f in flows)
+
+
+def test_cash_flows_dedup_cross_file_overlap():
+    """跨檔重疊期的同一筆現金流只算一次(去重骨架同 load());同檔同鍵多筆各自保留。"""
+    p1 = _write_csv(_CASH_CSV)
+    p2 = _write_csv(_CASH_CSV)          # 完全重疊的第二份 → 不該讓現金翻倍
+    flows = tr.load_cash_flows([p1, p2])
+    os.unlink(p1); os.unlink(p2)
+    assert len(flows) == 7, f"跨檔重疊去重後仍 7 筆,實得 {len(flows)}"
+
+
+def test_cash_position_unanchored_flags_unreliable():
+    """無現金餘額錨點 → csv_sum(假設開戶現金 0)+ reliable=False(CSV 漏 deposit 就偏,honesty 揭露)。"""
+    p = _write_csv(_CASH_CSV); flows = tr.load_cash_flows([p]); os.unlink(p)
+    cp = tr.cash_position(flows, held_mv=10000.0)
+    assert cp["source"] == "csv_sum" and cp["reliable"] is False
+    assert _approx(cp["balance"], 2612.50), cp["balance"]
+
+
+def test_cash_position_anchored_uses_anchor_plus_after():
+    """有錨點 → balance = 錨點餘額 + 錨點日之後的現金流(對付不完整 CSV);reliable=True,weight 正確。"""
+    p = _write_csv(_CASH_CSV); flows = tr.load_cash_flows([p]); os.unlink(p)
+    after = sum(f["amount"] for f in flows if f["date"] > dt.date(2024, 1, 15))
+    cp = tr.cash_position(flows, held_mv=10000.0,
+                          anchor={"as_of": "2024-01-15", "amount": 5000.0})
+    assert cp["source"] == "anchored" and cp["reliable"] is True
+    assert _approx(cp["balance"], 5000.0 + after), (cp["balance"], 5000.0 + after)
+    assert _approx(cp["weight"], cp["balance"] / (10000.0 + cp["balance"])), cp["weight"]
+
+
+def test_cash_position_recent_net_deposit_filters_by_prev_end():
+    """recent_net_deposit = prev_end 後的外部淨流入(deposit−withdrawal;息費股利不是外部本金);
+    prev_end 前的入金不算(給『這筆新入金該不該部署』判讀,只看本期)。"""
+    p = _write_csv(_CASH_CSV); flows = tr.load_cash_flows([p]); os.unlink(p)
+    cp = tr.cash_position(flows, held_mv=10000.0, prev_end="2024-01-15")
+    assert _approx(cp["recent_net_deposit"], -2000.0), cp["recent_net_deposit"]   # 1/5 存款不算,2/15 提款算
+    cp2 = tr.cash_position(flows, held_mv=10000.0)
+    assert _approx(cp2["recent_net_deposit"], 3000.0), cp2["recent_net_deposit"]  # 全期 5000−2000
+
+
+def test_cash_position_none_weight_when_denom_nonpositive():
+    """買入為主 + 無錨點 → csv_sum 可能為負(入金未完整記);持倉市值+現金≤0 → weight=None 降級,不誤報。"""
+    p = _write_csv(
+        "Symbol,Quantity,Price,Action,TradeDate,Amount,RecordType\n"
+        "NVDA,100,100.00,BUY,2024-01-10,-10000.00,Trade\n")
+    flows = tr.load_cash_flows([p]); os.unlink(p)
+    cp = tr.cash_position(flows, held_mv=5000.0)          # balance=-10000,denom=5000-10000<0
+    assert cp["weight"] is None, cp["weight"]
+    assert cp["reliable"] is False
+
+
+def test_cash_position_none_weight_when_unanchored_negative():
+    """無錨點 + 負現金(買入為主、入金未記全 → csv_sum 假設破裂):即使 denom>0,weight 也是垃圾 → None。
+    對比:有錨點的負現金 = 真融資(margin),weight 負有意義,照報(不誤殺真槓桿)。"""
+    flows = [dict(date=dt.date(2024, 1, 10), amount=-30000.0, kind="trade", currency="USD")]
+    cp = tr.cash_position(flows, held_mv=50000.0)         # 無錨點,balance=-30000,denom=20000>0
+    assert cp["weight"] is None, cp["weight"]             # 負現金無錨點 → 不報(不是 -1.5 之類垃圾)
+    cp2 = tr.cash_position(flows, held_mv=50000.0, anchor={"as_of": "2024-01-01", "amount": -5000.0})
+    assert cp2["weight"] is not None and cp2["weight"] < 0, cp2["weight"]  # 有錨點負現金=融資,照報負值
+
+
 # ─────────────────── 標準庫 runner(免 pytest 即可跑,與 test_sample_styles 一致)───────────────────
 
 def _main():

tests/test_tr_json_contract.py

@@ -43,6 +43,7 @@
     "n_held", "headline_dim", "headline_metric", "commitment", "metrics",
     "rule", "insufficient_data", "holdings",
     "currency_meta",                                    # #51/#129 PR-2a(optional 附加欄,單幣 USD 時內容多為 None)
+    "cash",                                             # #171 PR-1:帳戶現金地基(balance/weight/source/reliable/recent_net_deposit;None=未提供現金錨點)
     "problem_events", "problem_opportunities",          # #137 問題帳:事件規約 + Opportunity Check 快照
 }
 # SKILL Step 1「metrics:全 metric 快照」+ 對帳反查用鍵;收尾 CLI 另存 metrics_snapshot 全量快照
@@ -183,6 +184,15 @@ def main():
             ok(bool(trade_recap.CYCLE_ID_RE.match(cid) or trade_recap.CYCLE_ID_UNKNOWN_RE.match(cid)),
                f"cycle_id 格式合契約({cid})——單一事實源 = engine.CYCLE_ID_RE")
 
+        # ── 2a. #171 現金地基:state.cash 形狀 + 無 TR_CASH 錨點時降級為 csv_sum/不可信 ──
+        cash = st["cash"]
+        ok(isinstance(cash, dict) and set(cash.keys()) ==
+           {"balance", "weight", "source", "reliable", "recent_net_deposit"},
+           "state.cash 5 欄位齊(balance/weight/source/reliable/recent_net_deposit)", repr(cash)[:120])
+        ok(cash["source"] == "csv_sum" and cash["reliable"] is False,
+           "無 TR_CASH 錨點 → cash 降級 csv_sum + reliable=False(honesty 據此揭露,不冒充精確)",
+           repr(cash))
+
         # ── 2b. #162 接線:main flow 的 held 必須走 FIFO 剩餘,不是 positions() 的 avg cost ──
         # 單元層(test_engine_units)測的是 fifo_held 純函式;這段釘「main() 真的接上它」——
         # 攤平後部分賣出,state.holdings.cost 必須=FIFO 剩餘批成本;接線混回 avg cost 即紅。