Trend following: each asset against its own past

daily = cu.fetch_daily(cu.SP500_EXAMPLES, start="2018-01-01", end="2026-07-01")
db = h5i_db.Database(cu.fresh_db("alpha_trend_following"), create=True)
prices = daily.sort_by([("ts", "ascending"), ("symbol", "ascending")])
db.create_table("prices", prices.schema, time_column="ts", sort_key=["ts", "symbol"])
db.append("prices", prices, note="30 large caps, 2018-2026")
db.snapshot("prices-v1", tables=["prices"], note="Everything here reads this cut")
print(f"{prices.num_rows:,} rows, {daily.to_pandas()['symbol'].nunique()} names")
daily.to_pandas().head(3)

previous = sql_expr("lag(adj_close)").over(partition_by="symbol", order_by="ts")
past = sql_expr(f"lag(adj_close, {LOOKBACK})").over(partition_by="symbol", order_by="ts")
features = (
    db.table("prices", snapshot="prices-v1")
    .with_columns(previous=previous, past=past)
    .with_columns(
        ret=col("adj_close") / col("previous") - 1,
        trend=col("adj_close") / col("past") - 1,
    )
    .filter(col("ret").is_not_null() & col("trend").is_not_null())
)
scaled = (
    features.with_columns(
        vol=col("ret").rolling_std(VOL_WINDOW, order_by="ts", partition_by="symbol")
    )
    .with_columns(annual_vol=col("vol") * ANNUALIZE)
    .select(
        ts=col("ts"),
        symbol=col("symbol"),
        ret=col("ret"),
        trend=col("trend"),
        annual_vol=col("annual_vol"),
    )
    .sort(["ts", "symbol"])
)
panel = scaled.to_pandas().dropna()
print(f"{len(panel):,} symbol-days with a signal and a volatility estimate")
panel.head(3).set_index("ts").round(4)

signal = np.sign(panel["trend"])
names = panel.groupby("ts")["symbol"].transform("size")
panel["weight"] = (
    signal * (TARGET_VOL / np.sqrt(names)) / panel["annual_vol"]
).clip(-0.2, 0.2)

base = panel.pivot(index="ts", columns="symbol", values="weight").fillna(0.0)
returns = panel.pivot(index="ts", columns="symbol", values="ret").fillna(0.0)
unscaled = (base.shift(1).fillna(0.0) * returns).sum(axis=1)
print(f"realized volatility before the overlay: "
      f"{unscaled.std() * ANNUALIZE:.1%} against a {TARGET_VOL:.0%} target")

realized = unscaled.rolling(VOL_WINDOW).std() * ANNUALIZE
scale = (TARGET_VOL / realized).shift(1).clip(upper=3.0).fillna(0.0)
weights = base.mul(scale, axis=0)

held = weights.shift(1).fillna(0.0)  # decided yesterday, held today
gross = (held * returns).sum(axis=1)
turnover = (weights - weights.shift(1).fillna(0.0)).abs().sum(axis=1)
net = gross - turnover * COST_BPS / 10_000

print(f"realized volatility after   {gross.std() * ANNUALIZE:.1%}")
print(f"average gross exposure      {held.abs().sum(axis=1).mean():.2f}x")
print(f"average net exposure        {held.sum(axis=1).mean():+.2f}x")
print(f"average daily turnover      {turnover.mean():.3f} of book")
print(f"annual cost at {COST_BPS:.0f}bp         {turnover.mean() * 252 * COST_BPS / 10_000:.2%}")

fig, ax = plt.subplots(figsize=(9, 4))
ax.plot(held.index, held.sum(axis=1), linewidth=1.0)
ax.axhline(0, color="black", linewidth=0.8)
ax.set_title("Net exposure through time")
ax.set_xlabel("Date")
ax.set_ylabel("Sum of weights")
fig.tight_layout()

returns_table = pa.Table.from_pandas(
    pd.DataFrame({"ts": net.index, "ret": net.to_numpy()}), preserve_index=False
)
db.create_table("trend_returns", returns_table.schema, time_column="ts")
db.append("trend_returns", returns_table, note=f"{LOOKBACK}-day trend, {TARGET_VOL:.0%} vol target")

benchmark_table = pa.Table.from_pandas(
    pd.DataFrame({"ts": returns.index, "ret": returns.mean(axis=1).to_numpy()}),
    preserve_index=False,
)
db.create_table("universe_returns", benchmark_table.schema, time_column="ts")
db.append("universe_returns", benchmark_table, note="equal-weight universe")
db.snapshot("returns-v1", tables=["trend_returns", "universe_returns"])

strategy = quant.returns(db, "trend_returns", snapshot="returns-v1")
universe = quant.returns(db, "universe_returns", snapshot="returns-v1")
comparison = pd.DataFrame(
    {"trend (net)": strategy.stats(benchmark=universe), "universe": universe.stats()}
)
comparison.round(4)

print(f"target volatility   {TARGET_VOL:.1%}")
print(f"realized volatility {strategy.stats()['annual_volatility']:.1%}")
print(f"beta to the universe {strategy.stats(benchmark=universe)['beta']:+.2f}")

curve = strategy.equity_curve().to_pandas()
bench_curve = universe.equity_curve().to_pandas()
fig, ax = plt.subplots(figsize=(9, 4.5))
ax.plot(curve["ts"], curve["cumulative_return"], linewidth=1.6, label="trend, net of costs")
ax.plot(bench_curve["ts"], bench_curve["cumulative_return"], linewidth=1.2, label="equal-weight universe")
ax.axhline(0, color="black", linewidth=0.8)
ax.set_title(f"{LOOKBACK}-day time-series momentum at a {TARGET_VOL:.0%} volatility target")
ax.set_xlabel("Date")
ax.set_ylabel("Cumulative return")
ax.legend()
fig.tight_layout()

joined = pd.DataFrame({"strategy": net, "universe": returns.mean(axis=1)}).dropna()
buckets = pd.qcut(joined["universe"], 5, labels=["worst", "poor", "flat", "good", "best"])
by_bucket = joined.groupby(buckets, observed=True).agg(
    days=("strategy", "size"),
    universe=("universe", "mean"),
    strategy=("strategy", "mean"),
)
by_bucket["difference"] = by_bucket["strategy"] - by_bucket["universe"]
by_bucket.round(5)

monthly = joined.resample("ME").apply(lambda column: (1 + column).prod() - 1)
worst_months = monthly.nsmallest(6, "universe")
print("the six worst months for the universe:")
print((worst_months * 100).round(2).to_string())
print(f"\nstrategy beat the universe in {int((worst_months['strategy'] > worst_months['universe']).sum())} of 6")

def run_lookback(days: int) -> dict:
    lag = sql_expr(f"lag(adj_close, {days})").over(partition_by="symbol", order_by="ts")
    frame = (
        db.table("prices", snapshot="prices-v1")
        .with_columns(previous=previous, past=lag)
        .with_columns(
            ret=col("adj_close") / col("previous") - 1,
            trend=col("adj_close") / col("past") - 1,
        )
        .filter(col("ret").is_not_null() & col("trend").is_not_null())
        .with_columns(
            vol=col("ret").rolling_std(VOL_WINDOW, order_by="ts", partition_by="symbol")
        )
        .select(
            ts=col("ts"),
            symbol=col("symbol"),
            ret=col("ret"),
            trend=col("trend"),
            annual_vol=col("vol") * ANNUALIZE,
        )
        .sort(["ts", "symbol"])
        .to_pandas()
        .dropna()
    )
    count = frame.groupby("ts")["symbol"].transform("size")
    frame["weight"] = (
        np.sign(frame["trend"]) * (TARGET_VOL / np.sqrt(count)) / frame["annual_vol"]
    ).clip(-0.2, 0.2)
    b = frame.pivot(index="ts", columns="symbol", values="weight").fillna(0.0)
    r = frame.pivot(index="ts", columns="symbol", values="ret").fillna(0.0)
    raw = (b.shift(1).fillna(0.0) * r).sum(axis=1)
    overlay = (TARGET_VOL / (raw.rolling(VOL_WINDOW).std() * ANNUALIZE)).shift(1).clip(upper=3.0).fillna(0.0)
    w = b.mul(overlay, axis=0)
    turn = (w - w.shift(1).fillna(0.0)).abs().sum(axis=1)
    series = (w.shift(1).fillna(0.0) * r).sum(axis=1) - turn * COST_BPS / 10_000
    return {
        "lookback": days,
        "sharpe": float(series.mean() / series.std() * np.sqrt(252)),
        "annual turnover": float(turn.mean() * 252),
        "days": len(series),
    }


plateau = pd.DataFrame([run_lookback(days) for days in (63, 126, 189, 252, 315, 378)])
plateau.round(3)

fig, ax = plt.subplots(figsize=(9, 4))
ax.plot(plateau["lookback"], plateau["sharpe"], marker="o", linewidth=1.6)
ax.axhline(0, color="black", linewidth=0.8)
ax.set_title("Net Sharpe against trend lookback")
ax.set_xlabel("Lookback (trading days)")
ax.set_ylabel("Net Sharpe")
fig.tight_layout()

correlation = returns.corr().to_numpy()
off_diagonal = correlation[~np.eye(len(correlation), dtype=bool)]
print(f"average pairwise correlation of the universe: {off_diagonal.mean():.2f}")
print(f"effective independent bets (1/(1+(n-1)r)) * n: "
      f"{len(correlation) / (1 + (len(correlation) - 1) * off_diagonal.mean()):.1f} "
      f"of {len(correlation)} names")