Searching a strategy space without fooling yourself

daily = cu.fetch_daily(cu.SP500_EXAMPLES[:10], start="2020-01-01", end="2026-07-01")
frame = daily.to_pandas()
frame = frame[frame["ts"] >= pd.Timestamp("2022-01-01", tz="UTC")]
market = cu.make_equity_market(
    pa.Table.from_pandas(frame, preserve_index=False), spread_bps=4.0
)

db = h5i_db.Database(cu.fresh_db("04_searching_a_strategy_space"), create=True)
for name in ("instruments", "book_deltas", "trades"):
    table = market[name]
    db.create_table(name, table.schema, time_column="ts_init")
    db.append(name, table, note="4bp synthetic book")
db.snapshot("tape-v1", tables=["instruments", "book_deltas", "trades"], note="One tape for every trial")

closes = frame.pivot(index="ts", columns="symbol", values="close").sort_index()
sessions = closes.index
book = market["book_deltas"].to_pandas()
book["session"] = book["ts_init"].dt.floor("s").dt.tz_localize("UTC")
quote_at = {(row.session, row.instrument_id): row.ts_init for row in book.itertuples()}
print(f"{len(sessions):,} sessions, {closes.shape[1]} names")
closes.tail(3)

def crossover_signals(fast: int, slow: int) -> pa.Table:
    """Target one lot per name while the fast average leads the slow one."""
    averages = {
        "fast": closes.rolling(fast).mean(),
        "slow": closes.rolling(slow).mean(),
    }
    holding = (averages["fast"] > averages["slow"]).astype(float) * LOT
    # The last session has no session after it to execute in, so it cannot
    # carry a target. The one before it is set flat, which makes final cash the
    # whole answer instead of cash plus an open position to mark.
    holding = holding.iloc[slow:-1]
    holding.iloc[-1] = 0.0

    stamps, targets, names = [], [], []
    for session, row in holding.iterrows():
        index = sessions.get_loc(session)
        execution = sessions[index + 1]
        for symbol, target in row.items():
            stamps.append(quote_at[(execution, symbol)] + dt.timedelta(microseconds=1))
            targets.append(float(target))
            names.append(symbol)
    order = sorted(range(len(stamps)), key=lambda i: (stamps[i], names[i]))
    return backtest.target_positions(
        [stamps[i] for i in order],
        [targets[i] for i in order],
        instrument_id=[names[i] for i in order],
        tag=f"ma-{fast}-{slow}",
    )


CANDIDATES = [
    (fast, slow)
    for fast, slow in itertools.product((10, 20, 50), (50, 100, 200))
    if fast < slow
]
for fast, slow in CANDIDATES:
    table = crossover_signals(fast, slow)
    name = f"signals_{fast}_{slow}"
    db.create_table(name, table.schema, time_column="ts")
    db.append(name, table, note=f"MA {fast}/{slow}")
db.snapshot("plans-v1", tables=[f"signals_{f}_{s}" for f, s in CANDIDATES])
print(f"{len(CANDIDATES)} candidates, one signals table each")
db.table(f"signals_{CANDIDATES[0][0]}_{CANDIDATES[0][1]}").head(3).to_pandas()

def config(run_id: str, signals: str, **execution) -> backtest.BacktestConfig:
    return backtest.BacktestConfig(
        run_id=run_id,
        portfolio=backtest.PortfolioConfig(starting_cash=CAPITAL),
        data=backtest.DataConfig(signals=signals, snapshot="tape-v1"),
        execution=backtest.ExecutionConfig(
            fee_kind="proportional", fee_rate=execution.get("fee_rate", 0.0005)
        ),
        output=backtest.OutputConfig(equity_interval_nanos=86_400_000_000_000),
    )


leaderboard = []
runs = {}
for fast, slow in CANDIDATES:
    label = f"ma-{fast}-{slow}"
    result = backtest.execute(db, config(label, f"signals_{fast}_{slow}"))
    runs[label] = result
    leaderboard.append(
        {
            "candidate": label,
            "fills": result["fills"],
            "commissions": result["commissions"],
            "net": result["final_cash"] - CAPITAL,
        }
    )
board = pd.DataFrame(leaderboard).sort_values("net", ascending=False)
board.round(2)

before = backtest.trial_count(db)
again = backtest.execute(db, config("ma-20-100", "signals_20_100"))
after = backtest.trial_count(db)
print(f"cached          {again['cached']}")
print(f"trial count     {before} -> {after}")
print(f"fork reused     {again['fork'] == runs['ma-20-100']['fork']}")
print(f"same final cash {again['final_cash'] == runs['ma-20-100']['final_cash']}")
assert again["cached"] and after == before
assert again["fork"] == runs["ma-20-100"]["fork"]

best_label = board.iloc[0]["candidate"]
best_fast, best_slow = best_label.split("-")[1:]
edges = [sessions[i] for i in (0, len(sessions) // 3, 2 * len(sessions) // 3, len(sessions) - 1)]
folds = backtest.WalkForward.of(
    *[
        backtest.ValidationWindows(
            train=(edges[i].tz_localize(None), edges[i + 1].tz_localize(None)),
            holdout=(edges[i + 1].tz_localize(None), edges[i + 2].tz_localize(None)),
        )
        for i in range(len(edges) - 2)
    ]
)
study = backtest.study(
    db,
    study_id="cost-tolerance",
    base=config(f"study-{best_label}", f"signals_{best_fast}_{best_slow}"),
    parameters={"execution.fee_rate": backtest.Range(0.0, 0.004)},
    search=backtest.RandomSearch(trials=8, seed=5),
    validation=folds,
    selection=backtest.TopK(k=3, metric="final_cash"),
)
ranked = pd.DataFrame(study.ranked())
columns = [c for c in ("trial", "parameters", "train_median_final_cash",
                       "holdout_median_final_cash") if c in ranked.columns]
print(f"trials {len(study.trials)}, reached the holdout {len(study.selected)}")
ranked[columns].round(2)

with_holdout = sum(
    1 for row in study.trials if any(key.startswith("fold0_holdout_") for key in row)
)
print(f"{with_holdout} of {len(study.trials)} trials ever saw the holdout")
assert with_holdout == len(study.selected) <= 3

try:
    import optuna  # noqa: F401

    available = True
except ImportError:
    available = False
print(f"optuna installed: {available}")
print(f"trials recorded in this database so far: {backtest.trial_count(db)}")

basket = {label: result for label, result in list(runs.items())[:6]}
report = quant.basket_payload(
    db,
    basket,
    basket_id="ma-crossover search",
    panels=quant.PORTFOLIO_PANELS + ("equity",),
    snapshot="tape-v1",
)
print(f"runs      {report.totals['runs']}")
print(f"net       {report.totals['net']:,.2f}")
print(f"drawdown  {report.totals.get('max_drawdown', 0):.4f}")
print(f"panels    {', '.join(report.drawn)}")
if report.skipped:
    print(f"not drawn {[item['panel'] for item in report.skipped]}")
path = report.to_html("data/cache/ma-search-basket.html")
print(f"\nwrote {len(path):,} bytes of self-contained HTML")

print(f"strategy candidates run   {len(CANDIDATES)}")
print(f"execution trials run      {len(study.trials)}")
print(f"total trials in ledger    {backtest.trial_count(db)}")
print(f"best in-sample candidate  {best_label}")