"""
台股財報品質 8 個訊號的三項驗證：三率三升、毛利率新高、營益率轉正、存貨、資本支出、合約負債、EPS 連續成長、ROE 穩定

每個訊號做兩種版本：
  單獨版：符合條件的股票依分數取前 n 檔（n = 20／30／50），每月最後一個交易日換股、等權。
  組合版：疊在月營收動能策略（/strategies/monthly-revenue-momentum）上，
          符合條件的候選股優先入選（資本支出是「排除增幅最大的 20%」），持股 N = 8／10／12。

三項驗證（「贏」＝日夏普高於同期 0050 含息買進持有）：
  1. 參數高原：訊號參數 3 個值 × 持股數 3 個值，9 組裡至少 7 組要贏。
  2. 前後兩段：回測期間對半切，兩段各自要贏。
  3. 交易成本：sim() 預設手續費 0.1425%（買賣各一次）與證交稅 0.3%（賣出）。
組合版另要比底好：原參數日夏普高於同持股數的月營收動能單獨，而且 9 組裡至少 7 組高於同持股數的底。

股票池（單獨版）：上市櫃普通股（四碼數字，含已下市），排除金融保險與存託憑證，20 日平均成交金額至少 2,000 萬元。
財報：損益與現金流都是單季值，各公司實際公布財報那天才用得到（index_str_to_date）。
期間：2016-04-01 起（近 12 季的條件要等 2015 年報公布），合約負債 2019-07-01 起（2018 年起才全面揭露）。
0050 含息：etl:adj_close 還原價買進持有，不經 sim、不含成本。日夏普＝日報酬平均 ÷ 標準差 × √252，不扣無風險利率。

執行：安裝 finlab 後直接跑，資料需要時會自動引導登入，約 5 分鐘。
輸出：metrics.json（每組參數的結果）、data.csv（各版本原參數與 0050 含息的每日淨值）。
"""
import datetime
import json
import math

import numpy as np
import pandas as pd
import finlab
from finlab import data
from finlab.backtest import sim

START = "2016-04-01"
START_CONTRACT = "2019-07-01"
MIN_TURNOVER_VALUE = 2e7
HOLDINGS = [20, 30, 50]
BASE_HOLDINGS = [8, 10, 12]


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


# ---------------------------------------------------------------------------
# 1. 股票池、交易日與 0050 含息
# ---------------------------------------------------------------------------
adj_close = plain("etl:adj_close")
bench = adj_close["0050"].dropna()
categories = pd.DataFrame(data.get("security_categories")).set_index("stock_id")["category"]
EXCLUDED_CATEGORIES = {"金融保險業", "金融業", "存託憑證"}


def is_common_stock(sid):
    """4 位數普通股；已下市的公司沒有產業別，用代號排除金融（28xx）與存託憑證（91xx）。"""
    return (
        len(sid) == 4 and sid.isdigit() and not sid.startswith(("0", "28", "91"))
        and categories.get(sid) not in EXCLUDED_CATEGORIES
    )


stocks = [s for s in adj_close.columns if is_common_stock(s)]
days = adj_close.loc["2015-01-01":].index
month_id = days.to_period("M")
month_ends = days[np.append(month_id[1:] != month_id[:-1], True)]
avg_turnover_value = plain("price:成交金額").reindex(columns=stocks).rolling(20).mean().reindex(days)

NOT_QUALIFIED = -9e15


def on_days(quarterly):
    """季資料換成各公司實際公布日，展開到每個交易日。

    不符合條件（NaN）的季先填一個值：index_str_to_date 遇到 NaN 會沿用上一季的值，
    已經不符合條件的公司就會被補回來。
    """
    dated = pd.DataFrame(quarterly.astype(float).fillna(NOT_QUALIFIED).index_str_to_date())
    dated.index = pd.to_datetime(dated.index)
    dated = dated.reindex(columns=stocks)
    daily = dated.reindex(dated.index.union(days)).ffill().reindex(days)
    return daily.where(daily > NOT_QUALIFIED / 2)


# ---------------------------------------------------------------------------
# 2. 八個訊號：參數 → (符合條件, 排序分數, 分數小的優先?)
# ---------------------------------------------------------------------------
def features(name):
    return data.get(f"fundamental_features:{name}")


def statement(name):
    return data.get(f"financial_statement:{name}")


gross_margin = features("營業毛利率")
operating_margin = features("營業利益率")
net_margin = features("稅後淨利率")
eps = features("每股稅後淨利")
roe = features("ROE稅後")                          # 單季，不年化
revenue_4q = statement("營業收入淨額").rolling(4).sum()
inventory = statement("存貨")
total_assets = statement("資產總額")
capex_4q = (-statement("取得不動產_廠房及設備")).rolling(4).sum()   # 現金流量表是負數
contract_liability = statement("合約負債_流動")


def growth(x, lag):
    """比 lag 季前的變化率；lag 季前不是正數就不算。"""
    prev = x.shift(lag)
    return x / prev.where(prev > 0) - 1


def three_margins_up(k):
    """三率三升：毛利率、營益率、淨利率連續 k 季都高於去年同季，本季淨利率 > 0；依營益率年增（百分點）排序。"""
    up = gross_margin.rise(4) & operating_margin.rise(4) & net_margin.rise(4)
    ok = up.sustain(k) & (net_margin > 0)
    return ok, (operating_margin - operating_margin.shift(4)).where(ok), False


def gross_margin_high(quarters):
    """本季毛利率是近 quarters 季最高（高於前 quarters - 1 季）；依毛利率排序。"""
    ok = gross_margin > gross_margin.shift(1).rolling(quarters - 1).max()
    return ok, gross_margin.where(ok), False


def operating_turnaround(k):
    """本季營益率 > 0、去年同季 < 0，營益率連續 k 季年增；依本季營益率排序。"""
    ok = operating_margin.rise(4).sustain(k) & (operating_margin > 0) & (operating_margin.shift(4) < 0)
    return ok, operating_margin.where(ok), False


def inventory_to_sales(lag):
    """Abarbanell & Bushee (1998)：存貨 ÷ 近四季營收，比 lag 季前的變化率越低越好（存貨至少占總資產 5%）。"""
    ratio = inventory / revenue_4q.where(revenue_4q > 0)
    score = growth(ratio, lag).where(inventory >= 0.05 * total_assets)
    return score < 0, score, True


def capex_intensity(lag):
    """Titman, Wei & Xie (2004)：近四季資本支出 ÷ 近四季營收，比 lag 季前的變化率越低越好。"""
    intensity = capex_4q.where(capex_4q >= 0) / revenue_4q.where(revenue_4q > 0)
    return None, growth(intensity, lag), True


def contract_liability_growth(lag):
    """流動合約負債比 lag 季前的成長率越高越好（合約負債至少是近四季營收的 10%）。"""
    score = growth(contract_liability, lag).where(contract_liability >= 0.1 * revenue_4q)
    return score > 0, score, False


def eps_growth_streak(n):
    """單季 EPS 連續 n 季高於去年同季，本季與去年同季都賺錢；依本季 EPS 年增率排序。"""
    ok = eps.rise(4).sustain(n) & (eps > 0) & (eps.shift(4) > 0)
    return ok, (eps / eps.shift(4) - 1).where(ok), False


def stable_roe(n):
    """近 n 季每季 ROE 至少 3.75%（年化約 15%）；依近 n 季 ROE 標準差由低到高排序。"""
    ok = roe.rolling(n).min() >= 3.75
    return ok, roe.rolling(n).std().where(ok), True


SIGNALS = {
    # 名稱: (訊號, 標題, 參數名稱, 參數值, 原參數, 起始日)
    "three-margins-up": (three_margins_up, "三率三升", "連續季數", [1, 2, 3], 2, START),
    "gross-margin-high": (gross_margin_high, "毛利率創近 N 季新高", "N 季", [4, 8, 12], 8, START),
    "operating-turnaround": (operating_turnaround, "營益率由負轉正、連續改善", "連續季數", [1, 2, 3], 2, START),
    "inventory-to-sales": (inventory_to_sales, "存貨營收比下降", "比較幾季前", [2, 4, 8], 4, START),
    "capex-intensity": (capex_intensity, "資本支出強度增幅低", "比較幾季前", [2, 4, 8], 4, START),
    "contract-liability": (contract_liability_growth, "流動合約負債成長", "比較幾季前", [1, 2, 4], 2, START_CONTRACT),
    "eps-growth-streak": (eps_growth_streak, "EPS 連續 N 季年增", "N 季", [2, 4, 6], 4, START),
    "stable-roe": (stable_roe, "ROE 高且穩定", "N 季", [4, 8, 12], 8, START),
}

# ---------------------------------------------------------------------------
# 3. 單獨版：月底依分數取前 n 檔
# ---------------------------------------------------------------------------
turnover_at_month_end = avg_turnover_value.reindex(month_ends)
liquid_at_month_end = turnover_at_month_end > MIN_TURNOVER_VALUE


def month_end_order(score, ascending):
    """月底依分數排名次（流動性門檻內）；同分時成交金額大的優先。"""
    score = score.reindex(month_ends).where(liquid_at_month_end)
    rank = score.rank(axis=1, ascending=ascending, method="average")
    tie_break = turnover_at_month_end.where(score.notna()).rank(axis=1, ascending=False, pct=True) * 0.5
    return (rank + tie_break).rank(axis=1, method="first")


def single(signal, param, start):
    """回傳「持股數 n → 回測」的函式；名次只算一次，3 種持股數共用。"""
    _, score, ascending = signal(param)
    order = month_end_order(on_days(score), ascending).loc[start:]
    return lambda n, **cost: sim(order <= n, resample="M", upload=False, **cost)


# ---------------------------------------------------------------------------
# 4. 組合版：月營收動能（站上 20／60／120 日線、近 3 月平均營收 > 近 12 月，5 日動能前 N 檔，
#    每月 11 日換股、10% 停損），符合訊號的候選股排前面
# ---------------------------------------------------------------------------
close = data.get("price:收盤價")
monthly_revenue = data.get("monthly_revenue:當月營收")
trend = (close > close.average(20)) & (close > close.average(60)) & (close > close.average(120))
revenue_up = monthly_revenue.average(3) > monthly_revenue.average(12)
momentum = (close / close.shift() - 1).rolling(5).mean()
candidates = momentum[trend & revenue_up]


def to_close_grid(flag):
    return flag.astype(float).reindex(index=close.index, columns=close.columns).ffill().fillna(0) > 0.5


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


def combo(signal, param, start):
    """回傳「持股數 n → 回測」的函式；財報訊號只展開一次，3 種持股數共用。"""
    ok, score, _ = signal(param)
    if ok is None:  # 資本支出：排除強度增幅前 20% 的公司
        excluded = to_close_grid(on_days(score).rank(axis=1, pct=True) >= 0.8)
        ranked = candidates[~excluded]
    else:
        # 5 日平均漲跌幅在 ±0.1 以內，加 1 就讓符合訊號的候選股全部排在前面
        preferred = to_close_grid(on_days(ok) > 0.5)
        ranked = candidates + preferred.astype(float)
    return lambda n, **cost: revenue_momentum(ranked, n, start, **cost)


# ---------------------------------------------------------------------------
# 5. 績效與三項驗證
# ---------------------------------------------------------------------------
def active(curve):
    """從淨值第一次變動的前一天起算。"""
    curve = curve.dropna()
    first_move = (curve != curve.iloc[0]).argmax()
    return curve.iloc[max(first_move - 1, 0):]


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


def versus_0050(curve):
    s = stats(active(curve))
    b = stats(bench.loc[s["start"]:s["end"]])
    return {"strategy": s, "bench": b, "win": bool(s["sharpe"] > b["sharpe"])}


def halves(curve):
    curve = active(curve)
    mid = curve.index[0] + (curve.index[-1] - curve.index[0]) / 2
    return [versus_0050(curve.loc[:mid]), versus_0050(curve.loc[mid:])]


def avg_holdings(report):
    held = (report.position.fillna(0) != 0).sum(axis=1)
    return round(float(held[held > 0].mean()), 1)


def annual_turnover(report):
    """單邊年周轉率：每次換股權重變動絕對值總和的一半，加總後除以年數（不含停損賣出）。"""
    position = report.position.fillna(0)
    weight = position.div(position.abs().sum(axis=1).replace(0, np.nan), axis=0).fillna(0)
    curve = active(report.creturn)
    years = (curve.index[-1] - curve.index[0]).days / 365.25
    return round(float(weight.diff().abs().sum(axis=1).sum() / 2 / years), 2)


ZERO_COST = {"fee_ratio": 0, "tax_ratio": 0}
SINGLE_N, COMBO_N = 30, 10  # 原參數的持股數


def validate(build, signal, start, params, center_param, sizes, center_size):
    """9 組參數各跑一次；原參數另跑一次不扣成本。回傳 (結果, 原參數回測, 不扣成本對照)。"""
    grid = []
    for p in params:
        run = build(signal, p, start)
        for n in sizes:
            report = run(n)
            row = {"param": p, "n": n, "avg_holdings": avg_holdings(report), **versus_0050(report.creturn)}
            grid.append(row)
            if (p, n) == (center_param, center_size):
                base, center = row, report
                zero_cost = versus_0050(run(n, **ZERO_COST).creturn)
            print(build.__name__, p, n, f"sharpe {row['strategy']['sharpe']:.3f} vs 0050 {row['bench']['sharpe']:.3f}", flush=True)
    result = {
        "grid": grid,
        "grid_wins": sum(r["win"] for r in grid),
        "base": base,
        "halves": halves(center.creturn),
        "annual_turnover": annual_turnover(center),
    }
    result["pass"] = {
        "neighbors": result["grid_wins"] >= 7,
        "halves": all(h["win"] for h in result["halves"]),
        "cost": base["win"],
    }
    result["pass"]["all"] = all(result["pass"].values())
    return result, center, zero_cost


results = {"single": {}, "combo": {}, "base_alone": {}}
curves = {}

# 底策略單獨（組合版要比它好）
for start in (START, START_CONTRACT):
    rows = {}
    for n in BASE_HOLDINGS:
        report = revenue_momentum(candidates, n, start)
        rows[n] = {**versus_0050(report.creturn), "halves": halves(report.creturn)}
        if n == COMBO_N:
            curves[f"revenue-momentum-{start[:4]}"] = active(report.creturn)
            rows[n]["zero_cost"] = versus_0050(revenue_momentum(candidates, n, start, **ZERO_COST).creturn)
    results["base_alone"][start] = rows

for key, (signal, title, param_name, params, center, start) in SIGNALS.items():
    spec = {"param_name": param_name, "params": params, "center": center, "start": start}

    res, report, zero_cost = validate(single, signal, start, params, center, HOLDINGS, SINGLE_N)
    res.update(title=title, **spec, zero_cost=zero_cost)
    results["single"][key] = res
    curves[key] = active(report.creturn)

    res, report, zero_cost = validate(combo, signal, start, params, center, BASE_HOLDINGS, COMBO_N)
    base = results["base_alone"][start]
    above_base = sum(r["strategy"]["sharpe"] > base[r["n"]]["strategy"]["sharpe"] for r in res["grid"])
    res.update(title=f"月營收動能＋{title}", **spec, above_base=above_base, zero_cost=zero_cost)
    res["pass"]["beats_base"] = bool(res["base"]["strategy"]["sharpe"] > base[COMBO_N]["strategy"]["sharpe"] and above_base >= 7)
    res["pass"]["new_strategy"] = res["pass"]["all"] and res["pass"]["beats_base"]
    results["combo"][key] = res
    curves[f"{key}+revenue-momentum"] = active(report.creturn)

meta = {
    "run_date": str(datetime.date.today()),
    "finlab_version": finlab.__version__,
    "data_end": str(bench.index[-1].date()),
    "liquidity_filter_twd": MIN_TURNOVER_VALUE,
    "benchmark": "0050 含息買進持有（etl:adj_close，不含交易成本）",
    "sharpe": "日報酬平均 / 標準差 × √252，未扣無風險利率",
}
with open("metrics.json", "w", encoding="utf-8") as f:
    json.dump({"meta": meta, **results}, f, ensure_ascii=False, indent=1)

table = pd.DataFrame({k: v / v.iloc[0] for k, v in curves.items()})
bench_part = bench.loc[table.index[0]:]
table["0050含息"] = bench_part / bench_part.iloc[0]
table.round(6).to_csv("data.csv", index_label="date")

for group in ("single", "combo"):
    for key, res in results[group].items():
        b = res["base"]
        print(f"{group:6s} {res['title']}: 日夏普 {b['strategy']['sharpe']:.2f}（0050 {b['bench']['sharpe']:.2f}），"
              f"9 組贏 {res['grid_wins']} 組，三項 {'通過' if res['pass']['all'] else '沒過'}")