營收強勢動能瘋狗（開盤價進出、排除處置股）

"""營收強勢動能瘋狗（開盤價進出、排除處置股）：跑到最新資料的版本（/strategies/liquidity-risk-analysis-backtest 的數字來源）。

條件與 /blog/liquidity-risk-analysis-backtest 範例 1 降低流動性風險後的版本相同：
- 股價創 250 日新高
- 排除月營收年增連 3 個月低於 -10%，也排除近 12 個月有 8 個月年增超過 60%
- 連 3 個月月營收比近 12 個月最低值高出 25% 以上，且月增率連 3 個月大於 -40%
- 近 10 日平均成交量大於 200 張，排除處置股
- 符合條件的股票取近 10 日均量最小的 5 檔，單檔上限 1/3，每月換股，開盤價進出，
  停損 8%，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

STRATEGY_NAME = "營收強勢動能瘋狗（開盤價進出、排除處置股）"
HIGH_DAYS = 250  # 股價創新高回看交易日
TOP_N = 5        # 持股檔數


def build_report():
    close = data.get("price:收盤價")
    vol_ma = data.get("price:成交股數").average(10)
    rev = data.get("monthly_revenue:當月營收")
    rev_year_growth = data.get("monthly_revenue:去年同月增減(%)")
    rev_month_growth = data.get("monthly_revenue:上月比較增減(%)")
    not_disposal = data.get("etl:disposal_stock_filter")  # True 代表當天不是處置股

    buy = (
        (close == close.rolling(HIGH_DAYS).max())
        & ~(rev_year_growth < -10).sustain(3)
        & ~(rev_year_growth > 60).sustain(12, 8)
        & (rev.rolling(12).min() / rev < 0.8).sustain(3)
        & (rev_month_growth > -40).sustain(3)
        & (vol_ma > 200 * 1000)
        & not_disposal
    )
    buy = vol_ma * buy
    position = buy[buy > 0].is_smallest(TOP_N)  # 符合條件的股票裡成交量最小的幾檔
    return sim(position, resample="M", position_limit=1 / 3, stop_loss=0.08, trade_at_price="open",
               upload=False, name=STRATEGY_NAME)


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()
    bench_all = data.get("etl:adj_close")["0050"].dropna()
    report = build_report()

    # 從第一次有部位的那天起算，0050 含息用同一段期間
    creturn = report.creturn.dropna()
    first_move = (creturn != creturn.iloc[0]).argmax()
    creturn = creturn.iloc[max(first_move - 1, 0):]
    bench = bench_all.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/liquidity-risk-analysis-backtest/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()