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

"""SEA09 — Asia-session mean-reversion on 1h bars.

HYPOTHESIS: During the Asian session (00-08 UTC), fade extreme moves back toward
the session open. If price has moved far up from the session open, go short
(expecting reversion); if far down, go long. Session mean-reversion idea.

BAR LABELING (1h bars):
  - A bar labeled/timestamped at "01:00 UTC" closes at 01:00 UTC (covers 00:00-01:00).
  - Close[00:00 UTC] = the midnight bar close = prior day's last bar.
  - Close[08:00 UTC] = end of the Asia window.

CAUSAL DECISION:
  target[i] = position to hold DURING bar i+1 (decided with data <= close[i]).

  Asian session window: we want to hold a position during the bars from
  01:00 UTC to 08:00 UTC (bars closing at those hours cover 00:00-01:00 ... 07:00-08:00).

  To hold during the bar closing at h+1 UTC, we set target at bar closing at h UTC.
  So to be active during hours 01..08 UTC, we set target at hours 00..07 UTC.

  At bar[i] closing at h (00..07):
    - We know the session open = close of the bar at h=00 of the current day (midnight).
      If h > 0, this is already in the past and known. If h == 0, we use the current bar's
      close as the session open (we'll be entering the next bar at h=1 anyway,
      and we don't know the overnight move yet — so for h=0 we set target=0 to avoid
      a contamination: we'd be computing signal from the same bar we're deciding on).
      Actually at h=0 (midnight), we just know close[00:00] but don't yet know if there
      will be an extreme move — so the target for bar(h=1) set at bar(h=0) should compare
      close[00:00] vs itself = 0 move. We'll mark target=0 for this bar.
    - For h in {1..7}: session_open = close of the 00:00 bar of the same day.
      session_move = (close[i] - session_open) / session_open
      z-score of session_move vs historical distribution (rolling 30d) -> signal strength.
      target[i] = -sign(session_move) * |z| if |z| > threshold -> fade the move.

GRID (4 variants, 1 TF each = 4 * 2 assets = 8 backtests — within budget):
  A: simple sign-fade, no z-threshold (fade any move, binary direction)
  B: z-score fade, threshold=1.0 (only fade "significant" moves)
  C: z-score proportional (continuous weight proportional to -z)
  D: z-score proportional + vol-target

We only test 1h (this is an intraday hourly hypothesis).
Total: 4 variants × 1 TF × 2 assets = 8 backtests. Within budget.
"""

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 _build_asia_features(df: pd.DataFrame, z_win_days: int = 30):
    """
    For each 1h bar at index i:
    - Compute session_move[i] = (close[i] - session_open) / session_open
      where session_open = close of the 00:00 UTC bar of the SAME day.
    - Causal: session_open for day D is known from bar(h=0, day D) onward.
    - z-score of session_move vs rolling historical moves (causal).
    Returns (hour_arr, session_move_arr, z_arr).
    """
    dt = pd.to_datetime(df["datetime"], utc=True)
    close = df["close"].values.astype(float)
    n = len(df)
    hour_arr = dt.dt.hour.values
    date_arr = dt.dt.date.values

    # Build date -> index of the 00:00 bar (the "session open" for that date)
    # The 00:00 UTC bar closes at midnight, so date is the same calendar date.
    session_open_by_date = {}  # date -> close at 00:00 UTC
    for i in range(n):
        if hour_arr[i] == 0:
            session_open_by_date[date_arr[i]] = close[i]

    # Compute session_move for each bar in Asian session (h in 0..7)
    session_move = np.full(n, np.nan)
    for i in range(n):
        h = hour_arr[i]
        d = date_arr[i]
        if h in range(1, 8):  # h=1..7 (h=0 excluded: move relative to itself = 0, no signal)
            so = session_open_by_date.get(d, np.nan)
            if np.isfinite(so) and so > 0:
                session_move[i] = (close[i] - so) / so

    # Compute rolling z-score of session_move (causal, only using past observations)
    # We compute it only for the non-NaN values (within-session bars), treating them
    # as a time series. For z-scoring we use a rolling window of z_win_days * ~7 (bars per day
    # in session = 7 bars at h=1..7).
    session_move_series = pd.Series(session_move)
    roll_mean = session_move_series.rolling(z_win_days * 7, min_periods=14).mean()
    roll_std = session_move_series.rolling(z_win_days * 7, min_periods=14).std()
    z_arr = ((session_move_series - roll_mean) / roll_std.replace(0, np.nan)).values
    z_arr = np.nan_to_num(z_arr, nan=0.0)

    return hour_arr, session_move, z_arr


def target_simple_fade(df: pd.DataFrame) -> np.ndarray:
    """
    Variant A: Fade any Asia-session move (binary sign-based).
    target[i] = -sign(session_move[i]) if h in [1..7], else 0.
    Holds the position during bar i+1 (so exposure hours = 02..09 UTC closes).
    We restrict to h in [0..6] so we hold during [1..7] UTC.
    """
    hour_arr, session_move, _ = _build_asia_features(df)
    n = len(df)
    target = np.zeros(n)
    for i in range(n):
        h = hour_arr[i]
        # Set target at h=0..6 -> holds during h+1=1..7 UTC bar
        if h in range(0, 7) and np.isfinite(session_move[i]):
            target[i] = -np.sign(session_move[i]) if session_move[i] != 0 else 0.0
        # h=0: session_move is NaN (no move yet), so target stays 0 — flat at bar(h=1)
        # Actually let's re-check: session_move[h=0] is NaN (excluded range(1,8) above).
        # So for h=0, target=0 (flat) -> we don't take a position at the very first bar.
    return target


def target_zscore_threshold(df: pd.DataFrame) -> np.ndarray:
    """
    Variant B: Fade only when z-score of move exceeds 1.0 (i.e., "significant" extremes).
    target[i] = -sign(z) if |z| > 1.0 and h in [0..6], else 0.
    """
    hour_arr, _, z_arr = _build_asia_features(df)
    n = len(df)
    target = np.zeros(n)
    THRESHOLD = 1.0
    for i in range(n):
        h = hour_arr[i]
        if h in range(0, 7):
            z = z_arr[i]
            if abs(z) > THRESHOLD:
                target[i] = -np.sign(z)
    return target


def target_zscore_proportional(df: pd.DataFrame) -> np.ndarray:
    """
    Variant C: Continuous fade proportional to -z (clipped to [-1, 1]).
    target[i] = clip(-z / 2.0, -1, 1) for h in [0..6], else 0.
    Dividing by 2.0 so that a z=2 sigma move gives full unit position.
    """
    hour_arr, _, z_arr = _build_asia_features(df)
    n = len(df)
    target = np.zeros(n)
    for i in range(n):
        h = hour_arr[i]
        if h in range(0, 7):
            target[i] = float(np.clip(-z_arr[i] / 2.0, -1.0, 1.0))
    return target


def target_zscore_vol_targeted(df: pd.DataFrame) -> np.ndarray:
    """
    Variant D: Proportional z-score fade + vol-targeting (20% annual vol, 2x cap).
    """
    direction = target_zscore_proportional(df)
    return al.vol_target(direction, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)


if __name__ == "__main__":
    print("SEA09 — Asia-session mean-reversion on 1h bars")
    print("Grid: 4 variants × 1 TF (1h) × 2 assets = 8 backtests")
    print()

    # Variant A: simple sign fade
    rep_a = al.study_weights("SEA09-A-simple-fade", target_simple_fade, tfs=("1h",))
    print("=== Variant A: simple sign fade ===")
    print(al.fmt(rep_a))
    print()

    # Variant B: z-score threshold
    rep_b = al.study_weights("SEA09-B-zscore-threshold", target_zscore_threshold, tfs=("1h",))
    print("=== Variant B: z-score threshold (|z|>1.0) ===")
    print(al.fmt(rep_b))
    print()

    # Variant C: z-score proportional
    rep_c = al.study_weights("SEA09-C-zscore-proportional", target_zscore_proportional, tfs=("1h",))
    print("=== Variant C: z-score proportional ===")
    print(al.fmt(rep_c))
    print()

    # Variant D: z-score vol-targeted
    rep_d = al.study_weights("SEA09-D-zscore-vol-target", target_zscore_vol_targeted, tfs=("1h",))
    print("=== Variant D: z-score proportional + vol-target ===")
    print(al.fmt(rep_d))
    print()

    # Pick best by holdout Sharpe
    reps = [rep_a, rep_b, rep_c, rep_d]
    labels = ["A-simple-fade", "B-zscore-threshold", "C-zscore-proportional", "D-zscore-vol-target"]
    best = max(reps, key=lambda r: r["verdict"].get("best_holdout_sharpe", -9))
    best_label = labels[reps.index(best)]

    print(f"=== BEST CONFIG: {best_label} ===")
    print(al.fmt(best))
    print()
    print("JSON:", al.as_json(best))