90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
"""Live end-to-end probe of the event-driven PIT backtest engine (Task 8).
|
|
|
|
Uses fully PIT-complete fake inputs to prove the engine produces a reconciled,
|
|
accounting-consistent result — this is the "happy path" complement to the
|
|
fail-closed readiness gate already verified separately.
|
|
"""
|
|
import datetime as dt
|
|
from app.backtest_engine import run_event_backtest
|
|
from app.backtest_readiness import _factor_keys_required
|
|
|
|
|
|
def series(syms, days=120, jump=None, after_price=20.0):
|
|
s = dt.date(2026, 1, 1)
|
|
out = {}
|
|
for sym in syms:
|
|
bars = []
|
|
for i in range(days):
|
|
d = (s + dt.timedelta(days=i))
|
|
px = (after_price if jump and d >= jump else 10.0)
|
|
bars.append({"date": d.isoformat(), "adjusted_close": px})
|
|
out[sym] = {"bars": bars}
|
|
return out
|
|
|
|
|
|
class FStore:
|
|
def __init__(self, rels):
|
|
self.r = rels
|
|
|
|
def series(self, k):
|
|
return [{"released_at": t, "observed_at": t, "value": 1.0}
|
|
for t in self.r.get(k, [])]
|
|
|
|
|
|
class FLedger:
|
|
def symbols(self):
|
|
return ["A"]
|
|
|
|
def entries(self, s):
|
|
return [{"symbol": "A", "ex_date": "2026-02-15",
|
|
"per_share": 1.0, "estimate": False}]
|
|
|
|
|
|
rels = {k: ["2026-01-02T09:00:00+07:00"] for k in _factor_keys_required()}
|
|
|
|
|
|
class SC:
|
|
def list_ids(self):
|
|
return ["1"]
|
|
|
|
def _load_manifest(self):
|
|
return {"snapshots": {"1": {"retrieved_at": "2026-01-02T09:00:00+07:00"}}}
|
|
|
|
|
|
def pit_scorer(symbols, as_of=None):
|
|
out = {}
|
|
for i, sym in enumerate(symbols):
|
|
out[sym] = {
|
|
"combined": 1.0 / (i + 1), "is_dividend": True,
|
|
"dividend_yield": 3.0,
|
|
"pit_meta": {"pit": True, "partial_pit": False, "note": "test"},
|
|
}
|
|
return out
|
|
|
|
|
|
ser = series(["A", "B", "C"], 120, jump=dt.date(2026, 2, 1))
|
|
res = run_event_backtest(
|
|
start="2026-01-05", end="2026-04-30", capital=100000,
|
|
factor_store=FStore(rels), siamchart_store=SC(),
|
|
dividend_ledger=FLedger(), price_series=ser,
|
|
score_fn=pit_scorer, symbols=["A", "B", "C"],
|
|
)
|
|
d = res.to_dict()
|
|
print("rebalances:", d["rebalances"])
|
|
print("leakage_guard:", d["leakage_guard"])
|
|
print("accounting_reconciled:", d["accounting_reconciled"])
|
|
print("final_equity:", round(d["final_equity"], 2))
|
|
print("realized:", d["realized_trading_pnl"], "unrealized:", d["unrealized_trading_pnl"])
|
|
print("dividend_cash:", d["dividend_cash_received"], "dividend_recv:", d["dividend_receivable"])
|
|
print("fees:", d["transaction_costs"], "net_return:", round(d["net_return"], 4))
|
|
print("holdings:", [(h["symbol"], h["qty"]) for h in d["holdings"]])
|
|
|
|
assert d["accounting_reconciled"], "FAIL: not reconciled"
|
|
lhs = d["final_equity"] - 100000
|
|
rhs = (d["realized_trading_pnl"] + d["unrealized_trading_pnl"]
|
|
+ d["dividend_cash_received"] - d["transaction_costs"])
|
|
print(f" identity lhs={round(lhs,2)} rhs={round(rhs,2)}")
|
|
assert abs(lhs - rhs) < 1.0, "identity mismatch"
|
|
assert d["leakage_guard"] is True, "leakage_guard should be True with pit_meta"
|
|
print("PASS")
|