US short-term mean reversion (leveraged ETFs)

"""US short-term reversal ETF-timing strategy (FinLab).

Inside a positive QQQ trend regime, hold whichever of two leveraged growth
ETFs (TQQQ / TECL) lagged over the last 3 trading days. When the regime turns
risk-off, rotate into the strongest defensive asset (IEF / GLD / SHY).
Monthly rebalance, one holding at a time, 8% intramonth stop-loss.

Complete the AI-assisted setup flow at https://finlab.finance/en/setup before running this file.

    python strategy.py              # backtest
    python strategy.py --validate   # robustness checks from the article
"""

import sys

import finlab
from finlab import data
from finlab.backtest import sim
from finlab.dataframe import FinlabDataFrame

RISK_ASSETS = ["TQQQ", "TECL"]
DEFENSIVE_ASSETS = ["IEF", "GLD", "SHY"]


def build_position(reversal_days=3, trend_window=200):
    data.set_market("us_fund")
    close = data.get("us_fund_price:adj_close")[["QQQ"] + RISK_ASSETS + DEFENSIVE_ASSETS]

    # Regime filter: QQQ above its 200-day average AND positive 6-month momentum.
    qqq_trend = close["QQQ"] > close["QQQ"].rolling(trend_window, min_periods=trend_window // 2).mean()
    qqq_momentum = close["QQQ"] / close["QQQ"].shift(126) - 1
    risk_on = qqq_trend & (qqq_momentum > 0)

    # Reversal signal: rank the leveraged ETFs by NEGATIVE 3-day return,
    # so the recent laggard scores highest.
    reversal_score = FinlabDataFrame(-(close[RISK_ASSETS] / close[RISK_ASSETS].shift(reversal_days) - 1))

    # Defensive sleeve: strongest 3-month-minus-1-month momentum among IEF/GLD/SHY.
    defensive_score = FinlabDataFrame(
        close[DEFENSIVE_ASSETS].pct_change(63) - close[DEFENSIVE_ASSETS].pct_change(21)
    )

    risky_position = reversal_score.is_largest(1) & risk_on.to_frame().reindex(reversal_score.index).iloc[:, 0]
    defensive_position = (
        defensive_score.is_largest(1) & (~risk_on).to_frame().reindex(defensive_score.index).iloc[:, 0]
    )

    position = (
        risky_position.reindex(close.index).fillna(False).astype(float)
        .join(defensive_position.reindex(close.index).fillna(False).astype(float), how="outer")
        .fillna(0)
        .T.groupby(level=0).max().T
    )
    return position.loc["2016-01-01":]


def run_backtest(position, fee_ratio=None):
    data.set_market("us_fund")
    return sim(
        position,
        resample="M",
        name="us_short_term_reversal_etf",
        upload=False,
        fee_ratio=fee_ratio,    # None = finlab's US default (0 commission, 0 tax)
        trade_at_price="close",
        position_limit=1,
        stop_loss=0.08,
        touched_exit=True,
    )


# ---------------------------------------------------------------------------
# Robustness checks behind the article's validation section:
#     python strategy.py --validate
# "Beats QQQ" means a higher daily Sharpe than buying and holding QQQ
# (dividend-adjusted) over exactly the same dates.
# ---------------------------------------------------------------------------
COST_PER_SIDE = 0.001  # 0.1% per buy and per sell, for spread and slippage


def daily_sharpe(equity):
    returns = equity.pct_change().dropna()
    return returns.mean() / returns.std() * 252 ** 0.5


def sharpe_vs_qqq(equity):
    qqq = data.get("us_fund_price:adj_close")["QQQ"].reindex(equity.index).ffill()
    return daily_sharpe(equity), daily_sharpe(qqq)


def validate():
    # 1. Neighbouring parameters: the original value and one step either side.
    wins = 0
    for reversal_days in (2, 3, 4):
        for trend_window in (150, 200, 250):
            report = run_backtest(build_position(reversal_days=reversal_days, trend_window=trend_window))
            strategy, qqq = sharpe_vs_qqq(report.creturn)
            wins += strategy > qqq
            print(f"{reversal_days}-day laggard QQQ>{trend_window}d MA: Sharpe {strategy:.2f} vs QQQ {qqq:.2f}")
    print(f"{wins}/9 parameter sets beat QQQ")

    # 2. First half and second half of the backtest, each on its own.
    position = build_position()
    equity = run_backtest(position).creturn
    middle = equity.index[len(equity) // 2]
    for label, part in (("first half", equity.loc[:middle]), ("second half", equity.loc[middle:])):
        strategy, qqq = sharpe_vs_qqq(part)
        print(f"{label} {part.index[0].date()} to {part.index[-1].date()}: "
              f"Sharpe {strategy:.2f} vs QQQ {qqq:.2f}")

    # 3. Trading costs: finlab's US default is zero, so charge 0.1% per side.
    report = run_backtest(position, fee_ratio=COST_PER_SIDE)
    strategy, qqq = sharpe_vs_qqq(report.creturn)
    print(f"with 0.1% per side: CAGR {report.get_stats()['cagr']:.1%}, "
          f"Sharpe {strategy:.2f} vs QQQ {qqq:.2f}")


def main():
    finlab.login()  # opens a browser prompt; no token needed in the script
    if "--validate" in sys.argv:
        validate()
        return
    report = run_backtest(build_position())
    print(report.get_stats())
    report.display()


if __name__ == "__main__":
    main()