PythagorasGoal/scripts/research/alt/runs/SEA05.py

"""SEA05 — Intraday Momentum (1h)

HYPOTHESIS: On 1h bars, the sign of the morning session (00:00-11:59 UTC cumulative return)
predicts the afternoon session (12:00-23:59 UTC). Position taken at bar open of 12:00 UTC
and held through 23:59 UTC. Causal: morning return is fully known before entering at 12:00 close.

Implementation:
- Use 1h data only (the hypothesis requires intraday structure)
- For each day, compute morning cumulative return: product of (1+r) from 00:00 to 11:00 UTC (12 bars)
- At 12:00 UTC bar (i), compute signal from morning session (known at close[i-1] or earlier)
- Position = sign(morning_return) held for 12 bars (12:00 to 23:00 UTC)
- Vol-targeted continuous weights with vol_target(signal, df)

Grid: try 2 variants:
  A) raw sign (morning ret sign -> afternoon position)
  B) z-score of morning returns (magnitude matters -> stronger signal -> larger position)
"""
import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al
import numpy as np
import pandas as pd


def make_intraday_mom_signal(df: pd.DataFrame, use_zscore: bool = False, lookback_z: int = 20) -> np.ndarray:
    """
    For each 1h bar, compute an intraday momentum signal.

    Logic (causal):
    - Morning session = hours 0..11 UTC (12 bars per day)
    - At hour 12 (bar index where hour==12), the morning is complete
    - Signal = sign of morning cumulative return
    - Held for bars where hour in [12..23]
    - At hour 0 next day: flat (we re-evaluate)

    target[i] is set for bar i, evaluated with data up to close[i-1] for the morning.
    Actually: at bar with hour==12, close[i-1] is 11:00 close = last morning bar close.
    Morning return = close[11:00] / open[00:00] - 1 (for that day).
    """
    dt = df["datetime"]
    hour = dt.dt.hour

    # Compute log returns for each bar
    close = df["close"].values
    log_ret = np.zeros(len(df))
    log_ret[1:] = np.log(close[1:] / close[:-1])

    # Build daily morning cumulative return
    # For each bar at hour==12, sum log returns from hours 1..11 of same day
    # (hour 0 bar's return is from previous day's close to 00:00 close, we include it too)

    n = len(df)
    target = np.zeros(n)

    # We'll track morning cum-ret per day
    # Iterate bar by bar: accumulate morning, set signal at 12:00

    day_morning_cumret = 0.0
    morning_rets_history = []  # for z-score
    in_morning = False

    for i in range(n):
        h = hour.iloc[i]

        if h == 0:
            # Start of a new day: reset morning accumulator
            day_morning_cumret = 0.0
            in_morning = True

        if in_morning and h < 12:
            # Accumulate morning log return
            day_morning_cumret += log_ret[i]

        elif h == 12:
            # Morning complete, set position for afternoon
            in_morning = False

            if use_zscore and len(morning_rets_history) >= lookback_z:
                hist = np.array(morning_rets_history[-lookback_z:])
                mu = hist.mean()
                sigma = hist.std()
                if sigma > 1e-8:
                    z = (day_morning_cumret - mu) / sigma
                    # Clip to [-3, 3] and normalize
                    pos = np.clip(z / 2.0, -1.0, 1.0)
                else:
                    pos = 0.0
            else:
                # Simple sign
                pos = np.sign(day_morning_cumret)

            # Set target for this bar (12:00) and keep for afternoon
            # But we need to be careful: target[i] uses data up to close[i]
            # which is close of 12:00 bar. Actually we want to position DURING 12:00-23:00.
            # al.study_weights holds target[i] during bar i+1.
            # So: at bar i-1 (11:00), we know morning is done (close[i-1] = 11:00 close).
            # We should set target[i-1] to the signal so it's held during bar i (12:00 bar).
            # But that's complex. Instead: set target at i=12:00 bar using morning already
            # computed (morning is 00:00 to 11:00, all known before 12:00 bar opens).
            # The lib holds target[i] for bar i+1. So we set it at bar h==11 (last morning bar).
            # But we compute it here at h==12 for simplicity — let's adjust:
            # Actually set at h==11 (previous bar). We'll do a post-pass.

            # Store for z-score history
            morning_rets_history.append(day_morning_cumret)

            # We mark this as "12h signal" to be applied starting from 12:00 bar
            # Since lib shifts: target[i] held during bar i+1, we need target at i where h==11
            # We'll fix this in a second pass below; for now store in target[i]
            target[i] = pos

        elif h > 12:
            # Carry afternoon position forward
            target[i] = target[i-1]
        # else h in [1..11] or h==0: flat (0)

    # Shift the signal: target[i] where h==12 should be moved to h==11 bar
    # so that lib holds it during h==12 bar (bar i+1 from lib's perspective)
    # Find all bars where h==12, move signal to i-1 (h==11)
    afternoon_signal = np.zeros(n)
    i = 0
    while i < n:
        h = hour.iloc[i]
        if h == 12 and target[i] != 0:
            sig = target[i]
            # Set signal starting at i-1 (11:00 bar), lib holds it for bar i (12:00)
            # Actually we want to hold signal for bars 12..23
            # target[i-1] -> held during bar i (12:00) ✓
            # target[i]   -> held during bar i+1 (13:00) ✓
            # ...
            # target[i+10] -> held during bar i+11 (23:00) ✓
            # total: 12 bars (12:00-23:00)
            if i - 1 >= 0:
                afternoon_signal[i-1] = sig  # held during bar i (12:00)
            for k in range(i, min(i+11, n)):  # bars 12:00..22:00 -> held during 13:00..23:00
                afternoon_signal[k] = sig
            i += 12
        else:
            i += 1

    return afternoon_signal


def make_vol_targeted(df: pd.DataFrame, use_zscore: bool = False, lookback_z: int = 20) -> np.ndarray:
    """Intraday momentum with vol targeting."""
    raw_signal = make_intraday_mom_signal(df, use_zscore=use_zscore, lookback_z=lookback_z)
    # Vol-target: direction = sign(raw_signal), magnitude from vol_target
    direction = np.sign(raw_signal)
    w = al.vol_target(direction, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)
    return w


# Run the study with 2 variants on 1h only
print("=" * 60)
print("SEA05 — Intraday Momentum (1h)")
print("=" * 60)

# Variant A: simple sign, vol-targeted
print("\n--- Variant A: sign(morning_ret), vol-targeted ---")
rep_a = al.study_weights(
    "SEA05-A-sign",
    lambda df: make_vol_targeted(df, use_zscore=False),
    tfs=("1h",)
)
print(al.fmt(rep_a))
print("JSON:", al.as_json(rep_a))

# Variant B: z-score based magnitude, vol-targeted
print("\n--- Variant B: zscore(morning_ret), vol-targeted ---")
rep_b = al.study_weights(
    "SEA05-B-zscore",
    lambda df: make_vol_targeted(df, use_zscore=True, lookback_z=20),
    tfs=("1h",)
)
print(al.fmt(rep_b))
print("JSON:", al.as_json(rep_b))

# Pick best by min_asset_full_sharpe
best = rep_a if rep_a["verdict"]["best_full_sharpe"] >= rep_b["verdict"]["best_full_sharpe"] else rep_b
print("\n=== BEST CONFIG ===")
print(al.fmt(best))
print("JSON:", al.as_json(best))