US ETF rotation, conservative (QLD/SSO)

"""US ETF rotation: one regime signal, two risk levels (FinLab).

Both systems read the same risk-on/risk-off signal from QQQ (price above its
200-day moving average AND a positive 126-day return) and rebalance monthly:

  * "conservative" — rotates between the two strongest 2x leveraged equity
    ETFs (QLD / SSO, held 50/50) in risk-on, with a 6% stop.
  * "aggressive"   — holds the single strongest 3x leveraged growth ETF
    (TQQQ or TECL) in risk-on, with a 10% stop.

In risk-off, both park in the single strongest defensive ETF among
treasuries (IEF), gold (GLD), and cash-like short treasuries (SHY).

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

    python strategy.py              # backtest the profile chosen below
    python strategy.py --validate   # robustness checks from the article (both profiles)
"""
import sys

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

# Pick which version to run: "conservative" (QLD/SSO) or "aggressive" (TQQQ/TECL).
STRATEGY = "conservative"

PROFILES = {
    "conservative": {
        "risk_assets": ["QLD", "SSO"],   # 2x leveraged NASDAQ-100 / S&P 500
        "risk_holdings": 2,              # hold both, 50% each
        "risk_lookback": 126,            # rank by 6-month return
        "risk_skip": 0,
        "stop_loss": 0.06,               # 6% touched stop
    },
    "aggressive": {
        "risk_assets": ["TQQQ", "TECL"],  # 3x leveraged NASDAQ-100 / tech
        "risk_holdings": 1,               # hold only the strongest one
        "risk_lookback": 63,              # rank by 3-month minus 1-month return
        "risk_skip": 21,
        "stop_loss": 0.10,                # 10% touched stop
    },
}

DEFENSIVE_ASSETS = ["IEF", "GLD", "SHY"]  # treasuries / gold / cash-like (risk-off)


def build_position(profile, trend_window=200, momentum_window=126):
    data.set_market("us_fund")
    tickers = ["QQQ"] + profile["risk_assets"] + DEFENSIVE_ASSETS
    close = data.get("us_fund_price:adj_close")[tickers]

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

    # 2. Rank the risk ETFs by recent momentum (optionally skipping the last month).
    risk_score = close[profile["risk_assets"]].pct_change(profile["risk_lookback"])
    if profile["risk_skip"]:
        risk_score = risk_score - close[profile["risk_assets"]].pct_change(profile["risk_skip"])
    risk_score = FinlabDataFrame(risk_score)

    # 3. Rank defensive ETFs by 3-month-minus-1-month momentum.
    defensive_score = FinlabDataFrame(
        close[DEFENSIVE_ASSETS].pct_change(63) - close[DEFENSIVE_ASSETS].pct_change(21)
    )

    # 4. Risk-on: equal-weight the strongest risk ETFs. Risk-off: 1 defensive ETF.
    risky = risk_score.is_largest(profile["risk_holdings"]) & risk_on.to_frame().reindex(risk_score.index).iloc[:, 0]
    risky = risky.astype(float).div(risky.astype(float).sum(axis=1), axis=0).fillna(0)
    defensive = (
        defensive_score.is_largest(1) & (~risk_on).to_frame().reindex(defensive_score.index).iloc[:, 0]
    ).astype(float)

    position = (
        risky.reindex(close.index).fillna(0)
        .join(defensive.reindex(close.index).fillna(0), how="outer")
        .fillna(0)
        .T.groupby(level=0).max().T
    )
    return position.loc["2016-01-01":]


def run_backtest(profile, position, fee_ratio=None):
    data.set_market("us_fund")
    return sim(
        position,
        resample="M",                      # rebalance monthly
        position_limit=1,                  # max weight per asset (number of holdings
                                           # is set by is_largest(...) above)
        stop_loss=profile["stop_loss"],    # touched stop on the held ETF
        touched_exit=True,
        trade_at_price="close",
        fee_ratio=fee_ratio,               # None = finlab's US default (0 commission, 0 tax)
        upload=False,
    )


# ---------------------------------------------------------------------------
# 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(name):
    profile = PROFILES[name]
    print(f"--- {name}")

    # 1. Neighbouring parameters: the original value and one step either side.
    wins = 0
    for trend_window in (150, 200, 250):
        for momentum_window in (63, 126, 189):
            position = build_position(profile, trend_window, momentum_window)
            strategy, qqq = sharpe_vs_qqq(run_backtest(profile, position).creturn)
            wins += strategy > qqq
            print(f"QQQ>{trend_window}d MA {momentum_window}d momentum: "
                  f"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(profile)
    equity = run_backtest(profile, 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(profile, 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()  # finlab guides you through login automatically
    if "--validate" in sys.argv:
        for name in PROFILES:
            validate(name)
        return
    profile = PROFILES[STRATEGY]
    report = run_backtest(profile, build_position(profile))
    print(report.get_stats())
    report.display()


if __name__ == "__main__":
    main()