營收篩選創新高策略

"""營收篩選創新高策略：跑到最新資料的版本（/strategies/new-high-strategy 的數字來源）。

條件與 /blog/new-high-strategy 附的 strategy.py 相同，只是不截斷結束日：
- 20 日新高 + 3 個月平均營收創 12 個月新高 + 營收年增 > 10% + RSI > 50
- 20 日均量 > 300 張，依營收年增率選前 10 檔，單檔上限 12%
- 跟著月營收公布日換股，sim() 預設手續費 0.1425% 與證交稅 0.3%

統計口徑：日夏普 = 日報酬平均 / 標準差 × √252，不扣無風險利率；
0050 含息 = etl:adj_close 還原價買進持有，與策略同一段期間。
執行後會在目前目錄輸出 data.csv（淨值曲線）與 metrics.json（統計）。

    pip install finlab
    python strategy.py

finlab 需要資料時會自動引導登入，不需要在程式裡填 token。
"""

import json

import pandas as pd

import finlab
from finlab import data
from finlab.backtest import sim

START = "2015-01-01"
STRATEGY_NAME = "營收篩選創新高策略"
PRICE_HIGH_DAYS = 20
REV_GROWTH_MIN = 10
RSI_MIN = 50
MIN_VOLUME_SHARES = 300 * 1000
N_STOCKS = 10
POSITION_LIMIT = 0.12


def rsi_wilder(close: pd.DataFrame, period: int = 14) -> pd.DataFrame:
    """Compute Wilder RSI without relying on TA-Lib or pandas_ta."""
    delta = close.diff()
    gain = delta.clip(lower=0)
    loss = -delta.clip(upper=0)
    avg_gain = gain.ewm(alpha=1 / period, adjust=False, min_periods=period).mean()
    avg_loss = loss.ewm(alpha=1 / period, adjust=False, min_periods=period).mean()
    rs = avg_gain / avg_loss.replace(0, pd.NA)
    return 100 - (100 / (1 + rs))


def curve_stats(curve: pd.Series) -> dict:
    curve = curve / curve.iloc[0]
    daily = curve.pct_change().dropna()
    years = (curve.index[-1] - curve.index[0]).days / 365.25
    return {
        "cagr": curve.iloc[-1] ** (1 / years) - 1,
        "daily_sharpe": daily.mean() / daily.std() * 252 ** 0.5,
        "max_drawdown": (curve / curve.cummax() - 1).min(),
        "total_return": curve.iloc[-1] - 1,
    }


def main() -> None:
    finlab.login()

    close = data.get("price:收盤價")
    volume = data.get("price:成交股數")
    rev = data.get("monthly_revenue:當月營收")
    rev_growth = data.get("monthly_revenue:去年同月增減(%)")
    rsi = rsi_wilder(close, 14)

    volume_filter = volume.average(20) > MIN_VOLUME_SHARES
    price_new_high = close == close.rolling(PRICE_HIGH_DAYS).max()
    rev_ma3 = rev.average(3)
    rev_new_high = rev_ma3 == rev_ma3.rolling(12).max()
    rev_strong = rev_growth > REV_GROWTH_MIN
    high_rsi = rsi > RSI_MIN

    condition = volume_filter & price_new_high & rev_new_high & rev_strong & high_rsi
    position = rev_growth[condition].is_largest(N_STOCKS)

    # 月營收索引是公布截止日字串，轉成可交易日後沿用到下一次公布
    position = position.reindex(rev.index_str_to_date().index, method="ffill")
    position = position.loc[START:]

    report = sim(position, position_limit=POSITION_LIMIT, upload=False, name=STRATEGY_NAME)

    # 從第一次有部位的那天起算，0050 含息用同一段期間
    creturn = report.creturn.dropna()
    first_move = (creturn != creturn.iloc[0]).argmax()
    creturn = creturn.iloc[max(first_move - 1, 0):]
    bench = data.get("etl:adj_close")["0050"].dropna().loc[creturn.index[0]:creturn.index[-1]]

    strategy_stats = curve_stats(creturn)
    bench_stats = curve_stats(bench)
    holdings = (report.position.fillna(0) != 0).sum(axis=1)
    strategy_stats["avg_holdings"] = holdings[holdings > 0].mean()

    start, end = creturn.index[0].date(), creturn.index[-1].date()
    print(f"期間 {start} 至 {end}")
    for label, s in [("策略", strategy_stats), ("0050 含息", bench_stats)]:
        print(f"{label}：年化 {s['cagr']:.2%}，日夏普 {s['daily_sharpe']:.2f}，最大回撤 {s['max_drawdown']:.2%}")

    curves = pd.DataFrame({STRATEGY_NAME: creturn / creturn.iloc[0], "0050含息": bench / bench.iloc[0]})
    curves.index.name = "date"
    curves.round(6).to_csv("data.csv")
    metrics = {
        "window": f"{start}/{end}",
        "star": STRATEGY_NAME,
        "variants": {STRATEGY_NAME: {k: round(float(v), 6) for k, v in strategy_stats.items()}},
        "benchmark_0050": {k: round(float(v), 6) for k, v in bench_stats.items()},
        "run_date": str(pd.Timestamp.today().date()),
        "finlab_version": finlab.__version__,
        "generated_by": "static/blog/new-high-strategy/validation/strategy.py",
    }
    with open("metrics.json", "w", encoding="utf-8") as f:
        json.dump(metrics, f, ensure_ascii=False, indent=2)


if __name__ == "__main__":
    main()