月營收動能避開一字漲停打開

"""月營收動能避開一字漲停打開：跑到最新資料的版本（/strategies/limit-up-open-revenue-momentum 的數字來源）。

與 https://finlab.finance/blog/limit-up-open-revenue-momentum 同一份程式：
- 底：月營收動能選股原樣（/strategies/monthly-revenue-momentum）：站上 20／60／120 日均線、
  近 3 個月平均營收 > 近 12 個月平均營收、5 日平均漲跌幅最強 10 檔，每月 11 日換股、10% 停損
- 唯一新增：排除近 10 個交易日內「連續 2 天以上一字漲停後打開」的上市櫃普通股
  （四碼數字，不含 ETF 與存託憑證）
- 一字漲停：開盤＝最高＝最低＝收盤，且還原報酬 ≥ 9.5%（2015-06-01 前漲跌幅 7%，用 6.5%）；
  上市未滿 6 個交易日不算。打開＝連續一字漲停之後，第一天不是一字漲停（當天有成交）
- sim() 預設手續費 0.1425% 與證交稅 0.3%
和底策略的逐年、鄰近參數對照在研究程式 /blog/day-trading-limit-up-odd-lot-block-trade-backtest/research.py。

統計口徑：日夏普 = 日報酬平均 / 標準差 × √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 = "月營收動能避開一字漲停打開"
MIN_STREAK = 2      # 連續一字漲停天數：鄰近參數 1、2、3
AVOID_DAYS = 10     # 打開後排除幾個交易日：鄰近參數 5、10、20


def limit_up_opened(close, open_, high, low, adj_close):
    """連續 MIN_STREAK 天以上一字漲停之後，第一天不是一字漲停。"""
    adj_close = adj_close.reindex_like(close)
    ret = adj_close.pct_change(fill_method=None)
    limit = pd.Series(0.095, index=close.index)
    limit[limit.index < "2015-06-01"] = 0.065
    limit_up = (close == high) & ret.ge(limit, axis=0) & adj_close.shift(5).notna()
    one_word = limit_up & (open_ == high) & (low == high)

    # 連續一字漲停天數：累計次數減掉最近一次「不是一字」時的累計次數
    total = one_word.cumsum()
    streak = total - total.where(~one_word).ffill().fillna(0)
    return ~one_word & (streak.shift(1) >= MIN_STREAK) & close.notna()


def build_report():
    close = data.get("price:收盤價")
    rev = data.get("monthly_revenue:當月營收")

    # 1. 月營收動能（底策略原樣）
    rev_up = rev.average(3) > rev.average(12)
    trend = (close > close.average(60)) & (close > close.average(20)) & (close > close.average(120))
    momentum = (close / close.shift() - 1).rolling(5).mean()

    # 2. 近 AVOID_DAYS 個交易日內一字漲停打開過的股票
    opened = limit_up_opened(
        pd.DataFrame(close),
        pd.DataFrame(data.get("price:開盤價")),
        pd.DataFrame(data.get("price:最高價")),
        pd.DataFrame(data.get("price:最低價")),
        pd.DataFrame(data.get("etl:adj_close")),
    )
    avoid = opened.astype(float).rolling(AVOID_DAYS, min_periods=1).max() > 0
    common = [s for s in avoid.columns if len(s) == 4 and s.isdigit() and not s.startswith(("0", "91"))]
    avoid = avoid[common].reindex(columns=avoid.columns, fill_value=False)

    # 3. 每月 11 日換股、10% 停損，手續費與證交稅用 sim 預設
    position = momentum[trend & rev_up & ~avoid].is_largest(10).loc[START:]
    return sim(position, resample="M", resample_offset="11D", stop_loss=0.1, 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()
    report = build_report()

    # 從第一次有部位的前一個交易日起算，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()},
        "params": {"min_streak": MIN_STREAK, "avoid_days": AVOID_DAYS},
        "run_date": str(pd.Timestamp.today().date()),
        "finlab_version": finlab.__version__,
        "generated_by": "static/blog/limit-up-open-revenue-momentum/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()