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

"""TRD06 — Heikin-Ashi Trend Streak
HYPOTHESIS: Build HA candles; long while HA close > HA open (green streak), flat on color flip.
Also test vol-targeted variant and streak-length filter.

Configs tested (<=4 param sets, total backtests = 4 configs * 2 assets * 2 TFs = 16):
  1. Raw HA signal (long green, flat red) on 1d + 12h
  2. Vol-targeted HA signal
(We do 2 param sets * 2 TFs in study_weights call for a total of 8 runs x 2 assets = 16 cells)
"""
import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al
import numpy as np


def ha_candles(df):
    """Compute Heikin-Ashi OHLC causally.
    HA_close[i] = (open[i] + high[i] + low[i] + close[i]) / 4
    HA_open[i]  = (HA_open[i-1] + HA_close[i-1]) / 2
    This is causal: HA_open[i] uses only past HA values, HA_close[i] uses current bar data.
    """
    o = df["open"].values.astype(float)
    h = df["high"].values.astype(float)
    l = df["low"].values.astype(float)
    c = df["close"].values.astype(float)
    n = len(c)

    ha_o = np.zeros(n)
    ha_c = np.zeros(n)

    # HA_close is just the average of OHLC — uses current bar only, causal
    ha_c = (o + h + l + c) / 4.0

    # HA_open: bootstrapped from first bar, then recursively
    ha_o[0] = (o[0] + c[0]) / 2.0
    for i in range(1, n):
        ha_o[i] = (ha_o[i - 1] + ha_c[i - 1]) / 2.0

    return ha_o, ha_c


def trd06_base(df):
    """Long when HA candle is green (ha_close > ha_open), flat otherwise."""
    ha_o, ha_c = ha_candles(df)
    # signal: +1 when green, 0 when red/doji
    signal = np.where(ha_c > ha_o, 1.0, 0.0)
    return signal


def trd06_vt(df):
    """Vol-targeted version of TRD06: scale green signal by vol target."""
    ha_o, ha_c = ha_candles(df)
    direction = np.where(ha_c > ha_o, 1.0, 0.0)
    return al.vol_target(direction, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)


def trd06_streak2(df):
    """Long only when HA has been green for >= 2 consecutive bars (reduces noise)."""
    ha_o, ha_c = ha_candles(df)
    green = (ha_c > ha_o).astype(float)
    n = len(green)
    streak = np.zeros(n)
    cnt = 0
    for i in range(n):
        if green[i] > 0:
            cnt += 1
        else:
            cnt = 0
        streak[i] = cnt
    # long only when streak >= 2
    signal = np.where(streak >= 2, 1.0, 0.0)
    return signal


def trd06_streak2_vt(df):
    """Vol-targeted streak>=2 variant."""
    ha_o, ha_c = ha_candles(df)
    green = (ha_c > ha_o).astype(float)
    n = len(green)
    streak = np.zeros(n)
    cnt = 0
    for i in range(n):
        if green[i] > 0:
            cnt += 1
        else:
            cnt = 0
        streak[i] = cnt
    direction = np.where(streak >= 2, 1.0, 0.0)
    return al.vol_target(direction, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)


if __name__ == "__main__":
    print("=== TRD06: Heikin-Ashi Trend Streak ===\n")

    # Config 1: raw HA green/flat
    print("--- Config 1: Raw HA green signal (1d, 12h) ---")
    rep1 = al.study_weights("TRD06-base", trd06_base, tfs=("1d", "12h"))
    print(al.fmt(rep1))
    print("JSON:", al.as_json(rep1))

    print()

    # Config 2: vol-targeted HA
    print("--- Config 2: Vol-targeted HA (1d, 12h) ---")
    rep2 = al.study_weights("TRD06-VT", trd06_vt, tfs=("1d", "12h"))
    print(al.fmt(rep2))
    print("JSON:", al.as_json(rep2))

    print()

    # Config 3: streak>=2 filter
    print("--- Config 3: HA streak>=2 (1d only) ---")
    rep3 = al.study_weights("TRD06-streak2", trd06_streak2, tfs=("1d",))
    print(al.fmt(rep3))
    print("JSON:", al.as_json(rep3))

    print()

    # Config 4: streak>=2 vol-targeted
    print("--- Config 4: HA streak>=2 vol-targeted (1d only) ---")
    rep4 = al.study_weights("TRD06-streak2-VT", trd06_streak2_vt, tfs=("1d",))
    print(al.fmt(rep4))
    print("JSON:", al.as_json(rep4))

    # Summary: pick best config
    all_reps = [
        ("TRD06-base-1d", rep1, "1d"),
        ("TRD06-base-12h", rep1, "12h"),
        ("TRD06-VT-1d", rep2, "1d"),
        ("TRD06-VT-12h", rep2, "12h"),
        ("TRD06-streak2-1d", rep3, "1d"),
        ("TRD06-streak2-VT-1d", rep4, "1d"),
    ]

    print("\n=== SUMMARY ===")
    for label, rep, tf in all_reps:
        cell = next((c for c in rep["cells"] if c["tf"] == tf), None)
        if cell:
            print(f"{label:30s}: minFull={cell['min_asset_full_sharpe']:+.3f} "
                  f"minHold={cell['min_asset_holdout_sharpe']:+.3f} "
                  f"feeOK={cell['fee_survives']} grade={rep['verdict']['grade']}")