US free-cash-flow breadth timing (leveraged ETFs)

"""US free-cash-flow / balance-sheet strength regime strategy (FinLab).

How it works
------------
1. Build a liquid US stock universe: the top 500 names by 60-day average
   dollar volume, with price > $10 and dollar volume > $10M.
2. For every stock, check fundamentals (trailing twelve months, aligned to
   SEC filing dates): positive free cash flow, FCF margin above 5%, and
   debt-to-equity below 1.2, with price above its 200-day moving average.
3. Breadth gate: count what share of the universe passes. When more than 5%
   of liquid US stocks are cash-generating and in an uptrend - and QQQ itself
   is above its 200-day average with positive 6-month momentum - the market
   regime is "risk-on".
4. Execution: in risk-off, hold the strongest defensive asset
   (IEF / GLD / SHY). In risk-on, the 2 strongest leveraged growth ETFs
   (TQQQ / TECL) by 6-month momentum are added next to it, so the book is
   about 25% / 25% / 50% defensive. Monthly rebalance, 8% stop-loss.

Requires a FinLab account (free tier works): https://ai.finlab.tw

    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"]
START_DATE = "2016-01-01"


def build_position(breadth_threshold=0.05, trend_window=200):
    # -----------------------------------------------------------------------
    # 1. Liquid US universe (point-in-time, survivorship-safe)
    # -----------------------------------------------------------------------
    data.set_market("us")
    close = data.get("us_price:adj_close")
    raw_close = data.get("us_price:close")  # volume is not split-adjusted, so
    volume = data.get("us_price:volume")    # dollar volume uses the unadjusted close

    dollar_volume = (raw_close * volume).rolling(60, min_periods=20).mean()
    top_liquid = dollar_volume.is_largest(500)
    liquid_enough = dollar_volume > 10_000_000
    price_ok = raw_close > 10

    returns = close.pct_change()
    extreme_move = returns.abs() >= 0.50
    known_bad = extreme_move.rolling(252, min_periods=1).max().shift(1).fillna(False) > 0
    clean = ~known_bad.astype(bool)

    universe = top_liquid & liquid_enough & price_ok & clean

    # -----------------------------------------------------------------------
    # 2. Free-cash-flow / balance-sheet fundamentals (filing-date aligned)
    # -----------------------------------------------------------------------
    revenue = data.get("us_income_statement:revenue")
    ocf = data.get("us_cash_flow:operating_cash_flow")
    capex = data.get("us_cash_flow:capital_expenditure")
    total_debt = data.get("us_balance_sheet:total_debt")
    equity = data.get("us_balance_sheet:total_stockholders_equity")

    fcf = (ocf + capex).rolling(4).sum()          # trailing-twelve-month free cash flow
    ttm_revenue = revenue.rolling(4).sum()
    fcf_margin = fcf / ttm_revenue
    debt_to_equity = total_debt / equity
    trend = close > close.average(200)

    fcf_entry = universe & trend & (fcf > 0) & (fcf_margin > 0.05) & (debt_to_equity < 1.2)

    # -----------------------------------------------------------------------
    # 3. Breadth gate + market trend filter
    # -----------------------------------------------------------------------
    fund_close = data.get("us_fund_price:adj_close")[["QQQ"] + RISK_ASSETS + DEFENSIVE_ASSETS]

    fcf_breadth = fcf_entry.sum(axis=1) / universe.sum(axis=1)
    qqq_trend = fund_close["QQQ"] > fund_close["QQQ"].rolling(trend_window, min_periods=trend_window // 2).mean()
    qqq_momentum = fund_close["QQQ"] / fund_close["QQQ"].shift(126) - 1
    risk_on = (fcf_breadth > breadth_threshold) & (qqq_trend & (qqq_momentum > 0)).reindex(fcf_breadth.index)

    # -----------------------------------------------------------------------
    # 4. ETF rotation. Risk-off: 1 defensive ETF. Risk-on: the leveraged ETFs are added
    #    next to the defensive ETF (risk_on_daily is an object column after
    #    fillna, so ~risk_on_daily does not switch it off); the book ends up about
    #    25% TQQQ, 25% TECL and 50% defensive. The article's numbers are for this.
    # -----------------------------------------------------------------------
    risk_score = FinlabDataFrame(fund_close[RISK_ASSETS].pct_change(126))
    defensive_score = FinlabDataFrame(
        fund_close[DEFENSIVE_ASSETS].pct_change(63) - fund_close[DEFENSIVE_ASSETS].pct_change(21)
    )

    risk_on_daily = risk_on.reindex(fund_close.index).fillna(False)
    risky_position = risk_score.is_largest(2) & risk_on_daily.to_frame().reindex(risk_score.index).iloc[:, 0]
    risky_position = risky_position.astype(float).div(risky_position.astype(float).sum(axis=1), axis=0).fillna(0)
    defensive_position = (
        defensive_score.is_largest(1)
        & (~risk_on_daily).to_frame().reindex(defensive_score.index).iloc[:, 0]
    ).astype(float)

    position = (
        risky_position.reindex(fund_close.index).fillna(0)
        .join(defensive_position.reindex(fund_close.index).fillna(0), how="outer")
        .fillna(0)
        .T.groupby(level=0).max().T
    )
    return position.loc[START_DATE:]


def run_backtest(position, fee_ratio=None):
    # 5. Backtest (monthly rebalance, 8% stop-loss)
    data.set_market("us_fund")
    return sim(
        position,
        resample="M",
        name="fcf_balance_sheet_strength",
        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 breadth_threshold in (0.04, 0.05, 0.06):
        for trend_window in (150, 200, 250):
            report = run_backtest(build_position(breadth_threshold=breadth_threshold, trend_window=trend_window))
            strategy, qqq = sharpe_vs_qqq(report.creturn)
            wins += strategy > qqq
            print(f"breadth>{breadth_threshold:.2f} 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.
    equity = run_backtest(build_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(build_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:
        validate()
        return
    report = run_backtest(build_position())
    stats = report.get_stats()
    print(f"CAGR:           {stats['cagr']:.2%}")
    print(f"Monthly Sharpe: {stats['monthly_sharpe']:.2f}")
    print(f"Max drawdown:   {stats['max_drawdown']:.2%}")
    report.display()


if __name__ == "__main__":
    main()