"""月營收進階訊號回測：7 個訊號單獨選股、加在月營收動能上，各做三項驗證。

與 https://finlab.finance/blog/monthly-revenue-signals-backtest 同一份程式，文中所有數字都由它產生。

訊號（月表上算，index 是每月營收公告期限日，當成可取得日）：
  s2 標準化營收驚喜（Jegadeesh & Livnat 2006 的 SURGE 月版），回看 12／24／36 個月
  s3 年增率加速：本月年增率 − 前 k 個月平均，k 1／3／6
  s4 累計營收年增率：近 3 個月／年初至今／近 12 個月
  s5 年增由負轉正：本月年增 > 0、前 k 個月都 < 0，k 1／3／6
  s6 連續 N 個月年增，N 3／6／12
  s7 產業擴散：年增公司比例最高的前 K 個產業，K 3／5／8
  s8 單月營收創 N 個月新高，N 6／12／24
（訊號 1「營收公布早晚」沒有歷史公布時間，無法回測，見文章說明。）

單獨版：上市櫃普通股（排除金融保險、存託憑證，保留已下市股票），20 日平均成交金額 2,000 萬元以上，
每月 11 日換股、等權，持有分數最高的 20／30／50 檔。
組合版：月營收動能選股（站上 20／60／120 日線、近 3 月均營收 > 近 12 月、5 日動能前 N 檔、10% 停損），
符合訊號條件的候選股排前面，N 8／10／12。

三項驗證：贏＝日夏普（日報酬平均 / 標準差 × √252，不扣無風險利率）高於同期 0050 含息。
9 組鄰近參數至少 7 組贏、回測期中點切開兩段都贏、sim() 預設手續費與證交稅。
組合版另外要原參數與 9 組中至少 7 組高於同持股數的原版，才算新策略。

    pip install finlab
    python research.py

約 5 分鐘。輸出 metrics.json（每個訊號的 9 組、兩段、零成本對照，以及營收驚喜分五組、持股重疊、原版交易拆分、近 3／5 年日夏普等文章裡的延伸數字）與 data.csv（原參數月底淨值）。
finlab 需要資料時會自動引導登入，不需要在程式裡填 token。
"""

import json
import math

import numpy as np
import pandas as pd

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

START = "2015-01-01"
MIN_TURNOVER = 2e7
HOLDINGS, HOLDINGS0 = [20, 30, 50], 30  # 單獨版持股數，原參數 30 檔
BASE_N, BASE_N0 = [8, 10, 12], 10  # 組合版持股數，原參數 10 檔
MIN_WINS = 7  # 9 組至少幾組要贏
ZERO_COST = {"fee_ratio": 0, "tax_ratio": 0}


def plain(key):
    df = pd.DataFrame(data.get(key))
    df.index = pd.to_datetime(df.index)
    return df


# ---------------------------------------------------------------- 統計口徑
def active_window(cr):
    """從淨值第一次變動的前一個交易日起算。"""
    cr = cr.dropna()
    first_move = (cr != cr.iloc[0]).argmax()
    return cr.iloc[max(first_move - 1, 0):]


def stats(cr):
    cr = cr.dropna()
    cr = cr / cr.iloc[0]
    ret = cr.pct_change().dropna()
    years = (cr.index[-1] - cr.index[0]).days / 365.25
    return {
        "start": cr.index[0].strftime("%Y-%m-%d"),
        "end": cr.index[-1].strftime("%Y-%m-%d"),
        "cagr": round(float(cr.iloc[-1] ** (1 / years) - 1), 6),
        "sharpe": round(float(ret.mean() / ret.std() * math.sqrt(252)), 6),
        "mdd": round(float((cr / cr.cummax() - 1).min()), 6),
        "total_return": round(float(cr.iloc[-1] - 1), 6),
    }


def compare(cr, bench, trim=True):
    s = stats(active_window(cr) if trim else cr)
    b = stats(bench.loc[s["start"]:s["end"]])
    return {"strategy": s, "bench": b, "win": bool(s["sharpe"] > b["sharpe"])}


def halves(cr, bench):
    cr = active_window(cr)
    mid = cr.index[0] + (cr.index[-1] - cr.index[0]) / 2
    return [compare(part, bench, trim=False) for part in (cr.loc[:mid], cr.loc[mid:])]


def main():
    finlab.login()

    # ------------------------------------------------------------ 股票池與月表
    adj = plain("etl:adj_close")
    bench = adj["0050"].dropna()
    categories = pd.DataFrame(data.get("security_categories")).set_index("stock_id")["category"]
    excluded = {"金融保險業", "金融業", "存託憑證"}
    stocks = [s for s in adj.columns
              if len(s) == 4 and s.isdigit() and not s.startswith(("0", "28", "91"))
              and categories.get(s) not in excluded]
    days = adj.loc["2014-01-01":].index
    turnover = plain("price:成交金額").reindex(columns=stocks).rolling(20).mean().reindex(days)
    liquid = turnover > MIN_TURNOVER

    rev = plain("monthly_revenue:當月營收").reindex(columns=stocks)
    yoy = plain("monthly_revenue:去年同月增減(%)").reindex(columns=stocks).replace([np.inf, -np.inf], np.nan)
    ytd = plain("monthly_revenue:前期比較增減(%)").reindex(columns=stocks).replace([np.inf, -np.inf], np.nan)
    industry = categories.reindex(stocks)

    def to_days(mdf):
        """月表展開到每個交易日：每天沿用最近一個月的整列，該月沒有值就是 NaN，不會補回更早的月份。"""
        return mdf.astype(float).reindex(days, method="ffill")

    # ------------------------------------------------------------ 7 個訊號
    # 回傳 (條件月表或 None, 分數月表)；條件為 None 時，組合版用全市場分數前 30% 當條件
    def s2(lookback):
        d = rev - rev.shift(12)
        past = d.shift(1).rolling(lookback, min_periods=lookback)
        sd = past.std()
        return None, (d - past.mean()) / sd.where(sd > 0)

    def s3(k):
        return None, yoy - yoy.shift(1).rolling(k, min_periods=k).mean()

    def s4(window):
        if window == "ytd":
            return None, ytd
        prev = rev.shift(12).rolling(window, min_periods=window).sum()
        return None, rev.rolling(window, min_periods=window).sum() / prev.where(prev > 0) - 1

    def s5(k):
        q = (yoy > 0) & ((yoy.shift(1) < 0).rolling(k).sum() == k)
        return q, yoy.where(q)

    def s6(n):
        q = (yoy > 0).rolling(n).sum() == n
        return q, yoy.where(q)

    def s7(top_k):
        valid = yoy.notna()
        up = (yoy > 0) & valid
        n_ind = valid.T.groupby(industry).sum().T
        breadth = (up.T.groupby(industry).sum().T / n_ind).where(n_ind >= 5)
        # 擴散比例由高到低取前 K 個產業；同比例時檔數多的優先
        top = (breadth + n_ind / 1e6).rank(axis=1, ascending=False, method="first") <= top_k
        # 產業結果對回每檔股票，沒有產業分類的是 False
        q = top.reindex(columns=industry.values, fill_value=False).set_axis(industry.index, axis=1) & valid
        return q, yoy.where(q)

    def s8(n):
        q = (rev > 0) & (rev >= rev.rolling(n, min_periods=n).max())
        return q, yoy.where(q)

    signals = {
        "s2": (s2, "標準化營收驚喜", ("回看月數", [12, 24, 36]), 24),
        "s3": (s3, "營收年增率加速", ("對照前 k 月", [1, 3, 6]), 3),
        "s4": (s4, "累計營收年增率", ("累計區間", [3, "ytd", 12]), "ytd"),
        "s5": (s5, "營收年增由負轉正", ("先前負成長月數", [1, 3, 6]), 3),
        "s6": (s6, "營收連續 N 月年增", ("連續月數", [3, 6, 12]), 6),
        "s7": (s7, "產業營收擴散", ("前 K 個產業", [3, 5, 8]), 5),
        "s8": (s8, "單月營收創 N 月新高", ("N 個月", [6, 12, 24]), 12),
    }

    # ------------------------------------------------------------ 單獨版
    # 排名只跟參數 p 有關，每個 p 算一次，再取前 n 檔
    def single_rank(fn, p):
        _, score = fn(p)
        score = to_days(score).where(liquid)
        # 分數由高到低排名；同分時成交金額大的優先
        rank = score.rank(axis=1, ascending=False, method="average")
        tie = turnover.where(score.notna()).rank(axis=1, ascending=False, pct=True) * 0.5
        return (rank + tie).rank(axis=1, method="first").loc[START:]

    def single_sim(rank, n, **costs):
        return sim(rank <= n, resample="M", resample_offset="11D", upload=False, **costs)

    # ------------------------------------------------------------ 組合版（月營收動能）
    close = data.get("price:收盤價")
    mrev = data.get("monthly_revenue:當月營收")
    cand = ((close > close.average(20)) & (close > close.average(60)) & (close > close.average(120))
            & (mrev.average(3) > mrev.average(12)))
    momentum = (close / close.shift() - 1).rolling(5).mean()
    base_score = momentum[cand]

    def base_sim(score, n, **costs):
        position = score.is_largest(n).loc[START:]
        return sim(position, resample="M", resample_offset="11D", stop_loss=0.1, upload=False, **costs)

    def combo_score(fn, p):
        q, score = fn(p)
        if q is None:
            q = score.rank(axis=1, pct=True) >= 0.7
        bonus = to_days(q).reindex(index=close.index, columns=close.columns).ffill().fillna(0)
        return base_score + bonus

    # ------------------------------------------------------------ 跑三項驗證
    results, curves, base, kept = {}, {}, {}, {}
    for n in BASE_N:
        rep = base_sim(base_score, n)
        base[n] = {"cmp": compare(rep.creturn, bench), "halves": halves(rep.creturn, bench)}
        if n == BASE_N0:
            base_rep = rep
            curves["原版月營收動能"] = active_window(rep.creturn)
            base[n]["zero_cost"] = compare(base_sim(base_score, n, **ZERO_COST).creturn, bench)
    results["base_monthly_revenue_momentum"] = base

    kinds = (("single", single_rank, single_sim, HOLDINGS, HOLDINGS0),
             ("combo", combo_score, base_sim, BASE_N, BASE_N0))
    for key, (fn, title, (pname, pvals), p0) in signals.items():
        for kind, prepare, run, holdings, n0 in kinds:
            grid = []
            for p in pvals:
                prepared = prepare(fn, p)
                for n in holdings:
                    rep = run(prepared, n)
                    grid.append({"param": p, "holdings": n, **compare(rep.creturn, bench)})
                    if (p, n) == (p0, n0):
                        orig_rep, orig = rep, grid[-1]
                        zero_rep = run(prepared, n, **ZERO_COST)
            r = {
                "title": title if kind == "single" else f"月營收動能＋{title}",
                "param_name": pname, "params": pvals, "holdings": holdings, "original": [p0, n0],
                "grid": grid, "grid_wins": sum(g["win"] for g in grid),
                "halves": halves(orig_rep.creturn, bench),
                "zero_cost": compare(zero_rep.creturn, bench),
            }
            r["pass"] = {"neighbors": r["grid_wins"] >= MIN_WINS, "halves": all(h["win"] for h in r["halves"]),
                         "cost": orig["win"]}
            r["pass"]["all"] = all(r["pass"].values())
            if kind == "combo":
                r["above_base"] = sum(g["strategy"]["sharpe"] > base[g["holdings"]]["cmp"]["strategy"]["sharpe"]
                                      for g in grid)
                r["pass"]["new_strategy"] = (r["pass"]["all"] and r["above_base"] >= MIN_WINS
                                             and orig["strategy"]["sharpe"] > base[n0]["cmp"]["strategy"]["sharpe"])
            slug = key if kind == "single" else key + "c"
            results[slug] = r
            kept[slug] = orig_rep
            curves[r["title"]] = active_window(orig_rep.creturn)
            s = orig["strategy"]
            print(f"{slug:4s} {r['title']}：日夏普 {s['sharpe']:.2f}（0050 {orig['bench']['sharpe']:.2f}），"
                  f"9 組贏 {r['grid_wins']}，兩段 {[round(h['strategy']['sharpe'], 2) for h in r['halves']]}，"
                  f"通過 {r['pass']}", flush=True)

    # ------------------------------------------------------------ 文章裡的延伸數字
    def held(rep):
        return rep.position.fillna(0) != 0

    def turnover_rate(rep):
        """每次換股換掉的持股比例。"""
        h = held(rep)
        prev = h.shift(1, fill_value=False)
        n, m = h.sum(axis=1), (h.sum(axis=1) > 0) & (prev.sum(axis=1) > 0)
        return round(float(((h & ~prev).sum(axis=1)[m] / n[m]).mean()), 4)

    def versus_base(rep):
        a, b = held(rep), held(base_rep)
        idx, cols = a.index.intersection(b.index), a.columns.union(b.columns)
        a = a.reindex(index=idx, columns=cols, fill_value=False)
        b = b.reindex(index=idx, columns=cols, fill_value=False)
        na, nb = a.sum(axis=1), b.sum(axis=1)
        m = (na > 0) & (nb > 0)
        overlap = (a & b).sum(axis=1)[m] / pd.concat([na[m], nb[m]], axis=1).min(axis=1)
        corr = active_window(rep.creturn).pct_change().corr(active_window(base_rep.creturn).pct_change())
        return {"holdings_overlap": round(float(overlap.mean()), 4), "daily_return_corr": round(float(corr), 4),
                "turnover": turnover_rate(rep)}

    extra = {"base_turnover": turnover_rate(base_rep), "surge": versus_base(kept["s2"]),
             "new_high_combo": versus_base(kept["s8c"])}

    # 全期年化波動與近 3／5 年日夏普（營收驚喜、營收創新高組合版、原版、0050）
    # 0050 從營收驚喜第一次有部位的前一天起算
    surge = active_window(kept["s2"].creturn)
    named = {"surge": surge, "new_high_combo": active_window(kept["s8c"].creturn),
             "base": active_window(base_rep.creturn), "0050": bench.loc[surge.index[0]:]}
    end = bench.index[-1]
    extra["risk"] = {}
    for name, c in named.items():
        extra["risk"][name] = {
            "annual_vol": round(float(c.pct_change().std() * math.sqrt(252)), 4),
            "sharpe_last_3y": stats(c.loc[end - pd.DateOffset(years=3):])["sharpe"],
            "sharpe_last_5y": stats(c.loc[end - pd.DateOffset(years=5):])["sharpe"],
        }

    # 營收驚喜分五組：流動性股票池依 SURGE 百分位分組、組內等權
    pct = to_days(s2(24)[1]).where(liquid).rank(axis=1, pct=True).loc[START:]
    extra["surge_quintiles"] = {
        f"Q{q + 1}": compare(sim((pct > q / 5) & (pct <= (q + 1) / 5), resample="M", resample_offset="11D",
                                 upload=False).creturn, bench)["strategy"]
        for q in range(5)}

    # 原版的交易依進場訊號日營收是否創 12 個月新高分組
    new_high = to_days(s8(12)[0]).reindex(index=close.index).ffill().fillna(0) > 0.5
    trades = base_rep.get_trades()
    hit = [bool(new_high[sid].asof(d)) if sid in new_high.columns else False
           for sid, d in zip(trades["stock_id"].astype(str).str.split(" ").str[0], trades["entry_sig_date"])]
    split = trades.assign(new_high=hit).groupby("new_high")["return"]
    extra["base_trades_by_new_high"] = {
        str(k): {"n": int(g.size), "mean": round(float(g.mean()), 6), "median": round(float(g.median()), 6),
                 "win_rate": round(float((g > 0).mean()), 4)} for k, g in split}
    print(json.dumps(extra, ensure_ascii=False), flush=True)

    meta = {"run_date": str(pd.Timestamp.today().date()), "finlab_version": finlab.__version__,
            "data_end": str(bench.index[-1].date())}
    with open("metrics.json", "w", encoding="utf-8") as f:
        json.dump({**meta, "results": results, "extra": extra}, f, ensure_ascii=False, indent=1)

    df = pd.DataFrame(curves)
    df["0050含息"] = bench.reindex(df.index)
    df = df.loc[df.dropna(how="all").index[0]:]
    df = df / df.bfill().iloc[0]
    df = df.groupby(df.index.to_period("M")).tail(1)
    df.index.name = "date"
    df.round(6).to_csv("data.csv")


if __name__ == "__main__":
    main()