概览

# 1. Module-level ASSETS — the sandbox reads this before running anything
ASSETS = ["SPY", "BIL"]


# 2. Factory: precompute every indicator here, exactly once
def make_on_day(prices: pd.DataFrame):

    # 3. Daily callback: cheap lookups only
    def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
        engine.set_target_weights(dt, {"SPY": 1.0})

    # 4. Warmup date: the backtest starts here
    ready = prices.index[0]

    # 5. Always return the pair, in this order
    return on_day, ready

from __future__ import annotations

import numpy as np
import pandas as pd

from backtest.data import load_daily_prices
from backtest.engine import EngineConfig, PortfolioEngine
from backtest.metrics import compute_metrics

ASSETS = ["SPY", "QQQ", "TLT", "GLD", "BIL"]


def make_on_day(prices: pd.DataFrame):
    cols = [c for c in ASSETS if c in prices.columns]
    sma = prices[cols].rolling(200, min_periods=200).mean()
    state = {"last": None}

    def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
        if sma.loc[dt].isna().all():
            return
        key = (dt.year, dt.month)
        if state["last"] == key:
            return
        state["last"] = key
        long = [s for s in cols if prices.at[dt, s] > sma.at[dt, s]]
        w = 1.0 / len(long) if long else 0.0
        engine.set_target_weights(dt, {s: w for s in long})

    ready = sma.dropna(how="all").index.min()
    return on_day, ready