平均回帰・ペアトレードbotをBybitで実装する
OU過程・共和分・zスコア

"""pair_universe.py — 全銘柄の価格行列を作り、ペア候補を絞る
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  手順:
    1. 流動性のある銘柄に絞る（記事12のスクリーナーと同じ考え方）
    2. 同じ時刻グリッドに揃えた価格行列を作る
    3. 相関で粗く絞る（共和分検定は重いので、全組み合わせは避ける）

  重要:
    相関が高い = 共和分ではない。
    相関はあくまで「候補を絞るための安いフィルタ」として使う。
    最終判断は共和分検定（サンプル2）で行う。

  もう一つ重要:
    絞り込みの段階で検定回数を減らすことが、
    多重検定問題（サンプル7）の緩和にもなる。
"""
from __future__ import annotations

import time
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone

import numpy as np
import pandas as pd

from client import make_client
from kline_fetch import fetch_klines


@dataclass(frozen=True)
class UniverseConfig:
    category: str = "linear"
    quote: str = "USDT"
    interval: str = "60"
    days: int = 120
    min_turnover_24h: float = 20_000_000.0
    max_spread_bps: float = 5.0
    max_symbols: int = 60
    min_correlation: float = 0.70
    max_correlation: float = 0.995   # ほぼ同一の銘柄は除く
    sleep: float = 0.06


@dataclass
class PairCandidate:
    symbol_a: str
    symbol_b: str
    correlation: float
    n_observations: int

    @property
    def key(self) -> str:
        return f"{self.symbol_a}/{self.symbol_b}"

    def line(self) -> str:
        return (f"  {self.key:<26}"
                f"corr={self.correlation:>6.3f}  "
                f"n={self.n_observations}")


def liquid_symbols(http, cfg: UniverseConfig) -> list:
    """流動性のある銘柄を選ぶ。"""
    r = http.get_tickers(category=cfg.category)
    if r["retCode"] != 0:
        raise RuntimeError(f"get_tickers失敗: {r['retCode']} {r['retMsg']}")

    rows = []
    for t in r["result"]["list"]:
        symbol = t["symbol"]
        if not symbol.endswith(cfg.quote):
            continue
        bid, ask = float(t.get("bid1Price") or 0), float(t.get("ask1Price") or 0)
        if bid <= 0 or ask <= 0:
            continue
        mid = (bid + ask) / 2
        spread_bps = (ask - bid) / mid * 10_000
        turnover = float(t.get("turnover24h") or 0)

        if turnover < cfg.min_turnover_24h or spread_bps > cfg.max_spread_bps:
            continue
        rows.append((symbol, turnover, spread_bps))

    rows.sort(key=lambda x: -x[1])
    return [s for s, _, _ in rows[:cfg.max_symbols]]


def build_price_matrix(http, symbols: list,
                       cfg: UniverseConfig) -> pd.DataFrame:
    """同じ時刻グリッドに揃えた終値行列を作る。

    銘柄ごとに上場時期が違うため、欠損の扱いが重要。
    ffill で埋めると偽の定常性を作ってしまうので、
    共通期間だけを使う（dropna）。
    """
    start = (datetime.now(timezone.utc) - timedelta(days=cfg.days)) \
        .strftime("%Y-%m-%d")
    series = {}

    for i, symbol in enumerate(symbols, 1):
        try:
            bars = fetch_klines(http, symbol, interval=cfg.interval,
                                start=start, category=cfg.category,
                                verbose=False)
            if len(bars) < 100:
                continue
            series[symbol] = bars["close"]
        except Exception as e:
            print(f"  [skip] {symbol}: {e}")
        finally:
            time.sleep(cfg.sleep)

        if i % 10 == 0:
            print(f"  ... {i}/{len(symbols)}銘柄")

    if not series:
        return pd.DataFrame()

    df = pd.DataFrame(series)
    before = len(df)
    df = df.dropna()
    if before != len(df):
        print(f"  共通期間に揃えました: {before} → {len(df)}本")
    return df


def screen_pairs(prices: pd.DataFrame,
                 cfg: UniverseConfig = UniverseConfig()) -> list:
    """相関でペア候補を絞る。

    価格の相関ではなくリターンの相関を見る。
    価格の相関は「両方が上昇トレンド」というだけで高くなる（見せかけの相関）。
    """
    returns = np.log(prices / prices.shift(1)).dropna()
    corr = returns.corr()

    candidates = []
    symbols = list(prices.columns)

    for i in range(len(symbols)):
        for j in range(i + 1, len(symbols)):
            a, b = symbols[i], symbols[j]
            c = float(corr.loc[a, b])
            if cfg.min_correlation <= c <= cfg.max_correlation:
                candidates.append(PairCandidate(a, b, c, len(prices)))

    total_possible = len(symbols) * (len(symbols) - 1) // 2
    print(f"  総組み合わせ {total_possible:,} → "
          f"相関フィルタ後 {len(candidates):,}組")

    return sorted(candidates, key=lambda x: -x.correlation)


def save_prices(prices: pd.DataFrame,
                path: str = "./pair_prices.parquet") -> None:
    """価格行列を保存する。検定のたびに再取得しないため。"""
    prices.to_parquet(path)
    print(f"  保存: {path} ({prices.shape[0]}本 × {prices.shape[1]}銘柄)")


if __name__ == "__main__":
    cfg = UniverseConfig(days=120, max_symbols=40)
    c = make_client()

    print("=== 流動性のある銘柄を抽出 ===")
    symbols = liquid_symbols(c.http, cfg)
    print(f"  {len(symbols)}銘柄: {', '.join(symbols[:10])} ...")

    print("\n=== 価格行列を構築 ===")
    prices = build_price_matrix(c.http, symbols, cfg)
    if prices.empty:
        print("  データを取得できませんでした。")
        raise SystemExit(1)

    print(f"  {prices.shape[0]}本 × {prices.shape[1]}銘柄 "
          f"({prices.index[0]} 〜 {prices.index[-1]})")
    save_prices(prices)

    print("\n=== ペア候補の抽出 ===")
    candidates = screen_pairs(prices, cfg)

    print(f"\n  上位20組:")
    for candidate in candidates[:20]:
        print(candidate.line())

    print(f"\n※ ここで絞った {len(candidates)}組すべてを検定すると、")
    print(f"  有意水準5%なら偶然だけで約 {len(candidates) * 0.05:.0f}組が")
    print("  『有意』になる。サンプル7の補正が必須。")

"""cointegration.py — Engle-Granger と Johansen による共和分検定
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install statsmodels

参考: Engle, R.F. & Granger, C.W.J. (1987)
        "Co-integration and Error Correction", Econometrica 55(2)
      Johansen, S. (1991) "Estimation and Hypothesis Testing of
        Cointegration Vectors", Econometrica 59(6)

  2つの手法の違い:
    Engle-Granger: 2段階（回帰 → 残差のADF検定）。単純で解釈しやすい。
                   ただし「どちらを従属変数にするか」で結果が変わる。
    Johansen     : 多変量。順序に依存せず、3銘柄以上にも使える。
                   統計的に厳密だが、実装と解釈が重い。

  実務では両方を通し、両方が有意なペアだけを採用するのが安全。
"""
from __future__ import annotations

from dataclasses import dataclass, field

import numpy as np
import pandas as pd
from statsmodels.tsa.stattools import adfuller, coint
from statsmodels.tsa.vector_ar.vecm import coint_johansen


@dataclass
class CointResult:
    symbol_a: str
    symbol_b: str
    eg_pvalue: float = 1.0            # Engle-Granger の p値
    eg_statistic: float = 0.0
    adf_pvalue: float = 1.0           # 残差のADF検定
    johansen_trace: float = 0.0
    johansen_crit_95: float = 0.0
    hedge_ratio: float = 0.0
    n_observations: int = 0
    error: str = ""

    @property
    def eg_significant(self) -> bool:
        return self.eg_pvalue < 0.05

    @property
    def johansen_significant(self) -> bool:
        return self.johansen_trace > self.johansen_crit_95

    @property
    def both_significant(self) -> bool:
        return self.eg_significant and self.johansen_significant

    @property
    def key(self) -> str:
        return f"{self.symbol_a}/{self.symbol_b}"

    def line(self) -> str:
        if self.error:
            return f"  {self.key:<26}エラー: {self.error}"
        marks = ("EG" if self.eg_significant else "--") + \
                ("/JO" if self.johansen_significant else "/--")
        return (f"  {self.key:<26}"
                f"EG p={self.eg_pvalue:>7.5f}  "
                f"ADF p={self.adf_pvalue:>7.5f}  "
                f"JO {self.johansen_trace:>7.2f}/{self.johansen_crit_95:>6.2f}  "
                f"β={self.hedge_ratio:>8.4f}  [{marks}]")


def engle_granger(series_a: pd.Series, series_b: pd.Series) -> tuple:
    """Engle-Granger 検定。

    帰無仮説: 共和分していない
    p値が小さい = 共和分している証拠が強い
    """
    stat, pvalue, _ = coint(series_a, series_b, trend="c")
    return float(stat), float(pvalue)


def ols_hedge_ratio(series_a: pd.Series, series_b: pd.Series) -> tuple:
    """OLS でヘッジ比率と切片を求める。

    A = alpha + beta * B + residual
    """
    x = series_b.to_numpy()
    y = series_a.to_numpy()
    design = np.column_stack([np.ones(len(x)), x])
    coef, *_ = np.linalg.lstsq(design, y, rcond=None)
    alpha, beta = float(coef[0]), float(coef[1])
    residual = y - (alpha + beta * x)
    return beta, alpha, pd.Series(residual, index=series_a.index)


def residual_adf(residual: pd.Series) -> float:
    """残差の定常性を ADF 検定で確認する。

    帰無仮説: 単位根がある（非定常）
    p値が小さい = 定常である証拠が強い = 平均回帰する
    """
    result = adfuller(residual.dropna(), maxlag=None, autolag="AIC")
    return float(result[1])


def johansen_test(prices: pd.DataFrame,
                  det_order: int = 0, k_ar_diff: int = 1) -> tuple:
    """Johansen の trace 検定。

    戻り値: (trace統計量, 95%臨界値)
    trace > 臨界値 なら共和分関係が存在する。
    """
    result = coint_johansen(prices.to_numpy(), det_order, k_ar_diff)
    # r=0（共和分ベクトルが0本）の帰無仮説を検定する
    trace_stat = float(result.lr1[0])
    crit_95 = float(result.cvt[0, 1])     # [90%, 95%, 99%] の 95%
    return trace_stat, crit_95


def test_pair(prices: pd.DataFrame, symbol_a: str,
              symbol_b: str, use_log: bool = True) -> CointResult:
    """1ペアを両方の手法で検定する。

    use_log=True にすると対数価格を使う。
    価格水準の違い（BTCとDOGEでは桁が違う）の影響を減らせるため、
    実務では対数を使うことが多い。
    """
    result = CointResult(symbol_a, symbol_b)

    try:
        pair = prices[[symbol_a, symbol_b]].dropna()
        if len(pair) < 100:
            result.error = f"データ不足 ({len(pair)}本)"
            return result

        if use_log:
            pair = np.log(pair)

        a, b = pair[symbol_a], pair[symbol_b]
        result.n_observations = len(pair)

        # 1) Engle-Granger
        result.eg_statistic, result.eg_pvalue = engle_granger(a, b)

        # 2) ヘッジ比率と残差のADF
        beta, _, residual = ols_hedge_ratio(a, b)
        result.hedge_ratio = beta
        result.adf_pvalue = residual_adf(residual)

        # 3) Johansen
        result.johansen_trace, result.johansen_crit_95 = \
            johansen_test(pair)

    except Exception as e:
        result.error = f"{type(e).__name__}: {e}"

    return result


def test_all(prices: pd.DataFrame, candidates: list,
             use_log: bool = True, verbose: bool = True) -> list:
    """候補ペアをすべて検定する。"""
    results = []
    for i, candidate in enumerate(candidates, 1):
        result = test_pair(prices, candidate.symbol_a,
                           candidate.symbol_b, use_log)
        results.append(result)

        if verbose and i % 20 == 0:
            print(f"  ... {i}/{len(candidates)}組")

    return results


if __name__ == "__main__":
    from pair_universe import UniverseConfig, screen_pairs

    try:
        prices = pd.read_parquet("./pair_prices.parquet")
    except FileNotFoundError:
        print("pair_prices.parquet がありません。")
        print("先に pair_universe.py を実行してください。")
        raise SystemExit(1)

    print(f"価格行列: {prices.shape[0]}本 × {prices.shape[1]}銘柄\n")

    candidates = screen_pairs(prices, UniverseConfig())

    print(f"\n=== 共和分検定 ({len(candidates)}組) ===")
    results = test_all(prices, candidates)

    significant = [r for r in results if r.both_significant]
    eg_only = [r for r in results if r.eg_significant
               and not r.johansen_significant]

    print(f"\n  両方有意   : {len(significant)}組")
    print(f"  EGのみ有意 : {len(eg_only)}組")
    print(f"  有意でない : "
          f"{len(results) - len(significant) - len(eg_only)}組")

    print(f"\n=== 両方有意なペア（p値の小さい順・上位15） ===")
    for r in sorted(significant, key=lambda x: x.eg_pvalue)[:15]:
        print(r.line())

    expected_false = len(candidates) * 0.05
    print(f"\n!! 注意 !!")
    print(f"  {len(candidates)}組を有意水準5%で検定したので、")
    print(f"  共和分が全く無くても約 {expected_false:.0f}組が有意になる。")
    print(f"  実際に有意だったのは {len(significant)}組。")
    if len(significant) < expected_false * 1.5:
        print("  → 偶然の範囲内。このユニバースにペアは無いと考えるべき。")
    else:
        print("  → 偶然より多いが、個別ペアの信頼性は別問題。")
    print("  サンプル7の多重検定補正を必ず適用すること。")

"""hedge_ratio.py — OLS / TLS / Kalman Filter によるヘッジ比率推定
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

参考: Chan, E. (2013) "Algorithmic Trading: Winning Strategies
        and Their Rationale", Wiley（第3章 Kalman Filter による動的β）

  3つの方法:
    OLS    : A を B に回帰。単純だが「Bに誤差が無い」という非対称な仮定
    TLS    : 直交回帰。両方に誤差があると考える（対称）
    Kalman : βを状態変数として逐次更新。関係の変化に追随する

  実務での使い分け:
    - 検定・スクリーニングは OLS で十分
    - 実運用のβは Kalman を推奨（関係が変わるため）
    - ただし Kalman はパラメータ（観測ノイズ・状態ノイズ）の設定が難しい
"""
from __future__ import annotations

from dataclasses import dataclass, field

import numpy as np
import pandas as pd


@dataclass
class HedgeEstimate:
    method: str
    beta: float
    alpha: float
    spread: pd.Series
    beta_series: pd.Series | None = None    # Kalman のみ

    @property
    def spread_std(self) -> float:
        return float(self.spread.std())

    @property
    def spread_mean(self) -> float:
        return float(self.spread.mean())

    def line(self) -> str:
        beta_info = ""
        if self.beta_series is not None:
            beta_info = (f"  β範囲 [{self.beta_series.min():.4f}, "
                         f"{self.beta_series.max():.4f}]")
        return (f"  {self.method:<10}β={self.beta:>9.4f}  "
                f"α={self.alpha:>10.4f}  "
                f"σ(spread)={self.spread_std:>8.5f}{beta_info}")


def ols_hedge(a: pd.Series, b: pd.Series) -> HedgeEstimate:
    """通常最小二乗法。A = α + β·B + ε"""
    x, y = b.to_numpy(), a.to_numpy()
    design = np.column_stack([np.ones(len(x)), x])
    coef, *_ = np.linalg.lstsq(design, y, rcond=None)
    alpha, beta = float(coef[0]), float(coef[1])
    spread = pd.Series(y - (alpha + beta * x), index=a.index)
    return HedgeEstimate("OLS", beta, alpha, spread)


def tls_hedge(a: pd.Series, b: pd.Series) -> HedgeEstimate:
    """全最小二乗法（直交回帰）。

    OLSは「Bには誤差が無い」と仮定するため、
    AとBを入れ替えると β が 1/β にならない（非対称）。
    TLSは両方に誤差があると考えるので対称になる。
    """
    x, y = b.to_numpy(), a.to_numpy()
    x_mean, y_mean = x.mean(), y.mean()

    # 中心化したデータの主成分方向を求める
    data = np.column_stack([x - x_mean, y - y_mean])
    _, _, vt = np.linalg.svd(data, full_matrices=False)
    # 最小特異値に対応する方向が残差方向
    normal = vt[-1]
    if abs(normal[1]) < 1e-12:
        beta = 0.0
    else:
        beta = float(-normal[0] / normal[1])

    alpha = float(y_mean - beta * x_mean)
    spread = pd.Series(y - (alpha + beta * x), index=a.index)
    return HedgeEstimate("TLS", beta, alpha, spread)


def kalman_hedge(a: pd.Series, b: pd.Series,
                 delta: float = 1e-5,
                 observation_var: float = 1e-3) -> HedgeEstimate:
    """Kalman Filter による動的ヘッジ比率。

    状態: [alpha_t, beta_t]（時間とともに変化する）
    観測: A_t = alpha_t + beta_t * B_t + noise

    delta            : 状態遷移ノイズの大きさ。
                       大きいほどβが速く変化する（追随性↑ 安定性↓）
    observation_var  : 観測ノイズの分散

    delta の目安:
      1e-7  ほぼ固定β（OLSに近い）
      1e-5  緩やかに変化（推奨の出発点）
      1e-3  急激に変化（ノイズを追いかけすぎる）
    """
    x, y = b.to_numpy(), a.to_numpy()
    n = len(x)

    # 状態ノイズの共分散
    state_cov = delta / (1 - delta) * np.eye(2)

    # 初期状態は OLS の推定値
    initial = ols_hedge(a, b)
    state = np.array([initial.alpha, initial.beta])
    state_var = np.eye(2)

    alphas = np.zeros(n)
    betas = np.zeros(n)
    residuals = np.zeros(n)

    for t in range(n):
        # 予測ステップ
        state_var = state_var + state_cov

        # 観測行列
        h = np.array([1.0, x[t]])

        # 予測誤差
        predicted = float(h @ state)
        error = y[t] - predicted

        # 予測誤差の分散
        error_var = float(h @ state_var @ h) + observation_var

        # カルマンゲイン
        gain = (state_var @ h) / error_var

        # 更新ステップ
        state = state + gain * error
        state_var = state_var - np.outer(gain, h) @ state_var

        alphas[t] = state[0]
        betas[t] = state[1]
        residuals[t] = error       # 予測誤差がそのままスプレッド

    return HedgeEstimate(
        "Kalman", float(betas[-1]), float(alphas[-1]),
        pd.Series(residuals, index=a.index),
        beta_series=pd.Series(betas, index=a.index))


def rolling_ols_hedge(a: pd.Series, b: pd.Series,
                      window: int = 24 * 30) -> HedgeEstimate:
    """ローリングOLS。Kalmanより単純で挙動が読みやすい。"""
    betas = np.full(len(a), np.nan)
    alphas = np.full(len(a), np.nan)
    spread = np.full(len(a), np.nan)

    x, y = b.to_numpy(), a.to_numpy()

    for t in range(window, len(a)):
        xs = x[t - window:t]
        ys = y[t - window:t]
        design = np.column_stack([np.ones(window), xs])
        coef, *_ = np.linalg.lstsq(design, ys, rcond=None)
        alphas[t] = coef[0]
        betas[t] = coef[1]
        # そのバーのスプレッドは、直前までのβで計算する（未来を見ない）
        spread[t] = y[t] - (coef[0] + coef[1] * x[t])

    return HedgeEstimate(
        "RollingOLS", float(betas[-1]), float(alphas[-1]),
        pd.Series(spread, index=a.index).dropna(),
        beta_series=pd.Series(betas, index=a.index))


def compare_methods(prices: pd.DataFrame, symbol_a: str, symbol_b: str,
                    use_log: bool = True) -> dict:
    """4つの手法を比較する。"""
    pair = prices[[symbol_a, symbol_b]].dropna()
    if use_log:
        pair = np.log(pair)
    a, b = pair[symbol_a], pair[symbol_b]

    return {
        "OLS": ols_hedge(a, b),
        "TLS": tls_hedge(a, b),
        "RollingOLS": rolling_ols_hedge(a, b),
        "Kalman": kalman_hedge(a, b),
    }


if __name__ == "__main__":
    from statsmodels.tsa.stattools import adfuller

    prices = pd.read_parquet("./pair_prices.parquet")
    symbols = list(prices.columns)
    symbol_a, symbol_b = symbols[0], symbols[1]

    print(f"=== ヘッジ比率の比較 ({symbol_a} / {symbol_b}) ===\n")
    estimates = compare_methods(prices, symbol_a, symbol_b)

    for estimate in estimates.values():
        print(estimate.line())

    print("\n=== 残差の定常性（ADF p値。小さいほど定常） ===")
    for name, estimate in estimates.items():
        residual = estimate.spread.dropna()
        if len(residual) < 50:
            continue
        pvalue = adfuller(residual, autolag="AIC")[1]
        mark = "  ← 定常" if pvalue < 0.05 else ""
        print(f"  {name:<12}p={pvalue:.5f}{mark}")

    kalman = estimates["Kalman"]
    if kalman.beta_series is not None:
        betas = kalman.beta_series
        print(f"\n=== Kalman β の推移 ===")
        print(f"  初期 {betas.iloc[0]:.4f} → 最終 {betas.iloc[-1]:.4f}")
        print(f"  変動幅: {betas.max() - betas.min():.4f}")
        drift = abs(betas.iloc[-1] - betas.iloc[0]) / abs(betas.iloc[0]) \
            if betas.iloc[0] != 0 else 0
        print(f"  相対変化: {drift:.1%}")
        if drift > 0.3:
            print("  → βが大きく変化しています。")
            print("    固定βでは運用できません（関係が変質した可能性）")

    print("\n※ 実運用では Kalman か RollingOLS を使う。")
    print("  固定βは『過去全期間の平均的な関係』であり、")
    print("  現在の関係を表していない。")

"""ou_process.py — Ornstein-Uhlenbeck過程のパラメータ推定
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  OU過程:
    dX_t = θ(μ − X_t)dt + σ dW_t

    θ : 平均回帰速度（大きいほど速く戻る）
    μ : 長期平均
    σ : 拡散係数

  離散化して AR(1) として推定する:
    X_{t+1} − X_t = a + b·X_t + ε
    → θ = −b / Δt,  μ = −a / b,  σ = std(ε) / sqrt(Δt)

  ハーフライフ（乖離が半分に戻るまでの時間）:
    half_life = ln(2) / θ

  実務的な意味:
    ハーフライフが「1日」なら、乖離してから1日で半分戻る。
    保有期間の目安はハーフライフの1〜2倍。
    ハーフライフが長すぎる（数週間）ペアは、
    ファンディングコストで利益が消える。
"""
from __future__ import annotations

from dataclasses import dataclass

import numpy as np
import pandas as pd


@dataclass
class OuParams:
    theta: float          # 平均回帰速度（1バーあたり）
    mu: float             # 長期平均
    sigma: float          # 拡散係数
    half_life_bars: float
    r_squared: float
    n_observations: int
    bars_per_day: float = 24.0

    @property
    def half_life_hours(self) -> float:
        return self.half_life_bars * (24 / self.bars_per_day)

    @property
    def half_life_days(self) -> float:
        return self.half_life_bars / self.bars_per_day

    @property
    def stationary_std(self) -> float:
        """定常分布の標準偏差 = σ / sqrt(2θ)"""
        if self.theta <= 0:
            return float("nan")
        return self.sigma / np.sqrt(2 * self.theta)

    @property
    def is_mean_reverting(self) -> bool:
        return self.theta > 0 and np.isfinite(self.half_life_bars)

    def report(self) -> str:
        if not self.is_mean_reverting:
            return ("  平均回帰しません（θ <= 0）。"
                    "このスプレッドはランダムウォークかトレンドです")
        return "\n".join([
            f"  θ (回帰速度)      : {self.theta:.6f} /バー",
            f"  μ (長期平均)      : {self.mu:.6f}",
            f"  σ (拡散)          : {self.sigma:.6f}",
            f"  ハーフライフ      : {self.half_life_bars:.1f}バー "
            f"= {self.half_life_hours:.1f}時間 "
            f"= {self.half_life_days:.2f}日",
            f"  定常分布の標準偏差: {self.stationary_std:.6f}",
            f"  R²                : {self.r_squared:.4f}",
            f"  観測数            : {self.n_observations}",
        ])


def estimate_ou(spread: pd.Series, dt: float = 1.0,
                bars_per_day: float = 24.0) -> OuParams:
    """AR(1)回帰でOUパラメータを推定する。"""
    x = spread.dropna().to_numpy()
    if len(x) < 30:
        raise ValueError(f"データ不足 ({len(x)}件)")

    x_lag = x[:-1]
    dx = np.diff(x)

    design = np.column_stack([np.ones(len(x_lag)), x_lag])
    coef, *_ = np.linalg.lstsq(design, dx, rcond=None)
    a, b = float(coef[0]), float(coef[1])

    residual = dx - design @ coef
    ss_res = float((residual ** 2).sum())
    ss_tot = float(((dx - dx.mean()) ** 2).sum())
    r2 = 1 - ss_res / ss_tot if ss_tot > 0 else 0.0

    theta = -b / dt
    mu = -a / b if b != 0 else float(x.mean())
    sigma = float(residual.std(ddof=2)) / np.sqrt(dt)

    half_life = np.log(2) / theta if theta > 0 else float("inf")

    return OuParams(theta=theta, mu=mu, sigma=sigma,
                    half_life_bars=half_life, r_squared=r2,
                    n_observations=len(x), bars_per_day=bars_per_day)


def half_life_simple(spread: pd.Series) -> float:
    """ハーフライフだけを手早く計算する。

    実務ではこの1行があれば足りることも多い。
    """
    x = spread.dropna()
    lagged = x.shift(1).dropna()
    delta = (x - x.shift(1)).dropna()
    aligned = lagged.loc[delta.index]

    design = np.column_stack([np.ones(len(aligned)), aligned.to_numpy()])
    coef, *_ = np.linalg.lstsq(design, delta.to_numpy(), rcond=None)
    b = float(coef[1])
    return float(np.log(2) / -b) if b < 0 else float("inf")


def rolling_half_life(spread: pd.Series, window: int = 24 * 14) -> pd.Series:
    """ハーフライフの推移。平均回帰の性質が保たれているかを見る。"""
    values = []
    index = []

    for t in range(window, len(spread)):
        chunk = spread.iloc[t - window:t]
        try:
            values.append(half_life_simple(chunk))
        except Exception:
            values.append(np.nan)
        index.append(spread.index[t])

    return pd.Series(values, index=index)


def expected_holding_bars(params: OuParams, entry_z: float = 2.0,
                          exit_z: float = 0.0) -> float:
    """OU過程で entry_z から exit_z まで戻るのにかかる期待時間。

    E[T] ≈ ln(entry_z / exit_z) / θ  （exit_z > 0 の場合）
    exit_z = 0 なら理論上は無限になるので、0.1 で近似する。
    """
    if not params.is_mean_reverting:
        return float("inf")
    target = max(abs(exit_z), 0.1)
    if abs(entry_z) <= target:
        return 0.0
    return float(np.log(abs(entry_z) / target) / params.theta)


if __name__ == "__main__":
    from hedge_ratio import kalman_hedge, ols_hedge

    prices = pd.read_parquet("./pair_prices.parquet")
    symbols = list(prices.columns)
    symbol_a, symbol_b = symbols[0], symbols[1]

    pair = np.log(prices[[symbol_a, symbol_b]].dropna())
    a, b = pair[symbol_a], pair[symbol_b]

    print(f"=== OU過程の推定 ({symbol_a} / {symbol_b}) ===\n")

    for name, estimate in (("OLS", ols_hedge(a, b)),
                           ("Kalman", kalman_hedge(a, b))):
        print(f"[{name}によるスプレッド]")
        try:
            params = estimate_ou(estimate.spread, bars_per_day=24.0)
            print(params.report())

            print(f"\n  想定保有期間（z=2 → z=0.5）: "
                  f"{expected_holding_bars(params, 2.0, 0.5):.1f}バー")
        except ValueError as e:
            print(f"  推定できません: {e}")
        print()

    print("=== ハーフライフの推移 ===")
    spread = ols_hedge(a, b).spread
    hl = rolling_half_life(spread, window=24 * 14).dropna()
    if not hl.empty:
        finite = hl[np.isfinite(hl)]
        print(f"  中央値: {finite.median():.1f}バー "
              f"({finite.median() / 24:.2f}日)")
        print(f"  範囲  : {finite.min():.1f} 〜 {finite.max():.1f}バー")
        unstable = (~np.isfinite(hl)).mean()
        print(f"  非定常だった期間: {unstable:.1%}")
        if unstable > 0.3:
            print("  → 3割以上の期間で平均回帰していません。")
            print("    このペアは安定した関係を持っていません")

    print("\n【判断の目安】")
    print("  ハーフライフ 〜1日   : 短期。取引回数が多く手数料負けしやすい")
    print("  ハーフライフ 1〜5日  : 実用的な範囲")
    print("  ハーフライフ 5日〜   : ファンディングコストが効く（記事10）")
    print("  ハーフライフ = inf   : 平均回帰しない。取引すべきでない")

"""zscore_strategy.py — zスコアによるエントリー・エグジット設計
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  z = (spread − rolling_mean) / rolling_std

  規則:
    z > +entry  → スプレッドが高すぎる → A売り / B買い
    z < −entry  → スプレッドが安すぎる → A買い / B売り
    |z| < exit  → 決済
    |z| > stop  → 損切り（関係が壊れた可能性）

  閾値の決め方:
    - ローリング窓はハーフライフの3〜5倍が目安
      短すぎると平均が乖離を追いかけてしまい、zが0に張り付く
    - entry は 2.0 が定番。ただし取引回数とのトレードオフ
    - stop は entry の 1.5〜2倍。これを超えたら共和分の崩壊を疑う

  重要: rolling_mean / rolling_std は必ず shift(1) する。
        当日の値を含めると未来を見たことになる（記事31）。
"""
from __future__ import annotations

from dataclasses import dataclass, field

import numpy as np
import pandas as pd


@dataclass(frozen=True)
class ZConfig:
    lookback: int = 24 * 5        # zスコアの推定窓
    entry_z: float = 2.0
    exit_z: float = 0.5
    stop_z: float = 4.0
    max_holding_bars: int = 24 * 10
    fee_bps_per_leg: float = 5.5   # テイカー片道（往復2レッグで4回）
    funding_bps_per_day: float = 0.0   # 両建てなら概ね相殺されるが要検証


def compute_zscore(spread: pd.Series, lookback: int) -> pd.Series:
    """ローリングzスコア。必ず shift(1) して未来を見ない。"""
    mean = spread.rolling(lookback).mean().shift(1)
    std = spread.rolling(lookback).std().shift(1)
    return ((spread - mean) / std.replace(0, np.nan))


@dataclass
class Trade:
    entry_index: int
    entry_time: object
    direction: int          # +1: スプレッド買い(A買いB売り), -1: その逆
    entry_z: float
    entry_spread: float
    exit_index: int = -1
    exit_time: object = None
    exit_z: float = 0.0
    exit_spread: float = 0.0
    exit_reason: str = ""

    @property
    def holding_bars(self) -> int:
        return self.exit_index - self.entry_index if self.exit_index >= 0 else 0

    def gross_return(self) -> float:
        """スプレッドの変化による粗利（対数スプレッド前提）。"""
        return (self.exit_spread - self.entry_spread) * self.direction


@dataclass
class BacktestResult:
    trades: list = field(default_factory=list)
    config: ZConfig = field(default_factory=ZConfig)

    @property
    def n_trades(self) -> int:
        return len(self.trades)

    def net_returns(self) -> list:
        """手数料込みのリターン。

        1往復で 4回のテイカー約定（エントリー2レッグ + 決済2レッグ）。
        """
        cost = self.config.fee_bps_per_leg / 10_000 * 4
        out = []
        for t in self.trades:
            funding = (self.config.funding_bps_per_day / 10_000
                       * t.holding_bars / 24)
            out.append(t.gross_return() - cost - funding)
        return out

    def summary(self) -> dict:
        if not self.trades:
            return {"取引数": 0}

        net = np.array(self.net_returns())
        gross = np.array([t.gross_return() for t in self.trades])
        holding = np.array([t.holding_bars for t in self.trades])

        wins = net[net > 0]
        losses = net[net <= 0]

        equity = np.cumsum(net)
        peak = np.maximum.accumulate(equity)
        max_dd = float((equity - peak).min()) if len(equity) else 0.0

        reasons: dict = {}
        for t in self.trades:
            reasons[t.exit_reason] = reasons.get(t.exit_reason, 0) + 1

        return {
            "取引数": len(self.trades),
            "粗利合計": float(gross.sum()),
            "純利合計": float(net.sum()),
            "手数料負担": float(gross.sum() - net.sum()),
            "勝率": len(wins) / len(net),
            "平均利益": float(wins.mean()) if len(wins) else 0.0,
            "平均損失": float(losses.mean()) if len(losses) else 0.0,
            "平均保有バー": float(holding.mean()),
            "最大DD": max_dd,
            "決済理由": reasons,
        }


def backtest_zscore(spread: pd.Series,
                    cfg: ZConfig = ZConfig()) -> BacktestResult:
    """zスコア戦略のバックテスト（スプレッド単位の損益）。

    実際の損益は建玉額に依存するが、
    ここでは「スプレッドがどれだけ取れたか」を評価する。
    通貨建ての評価は記事31のバックテスト基盤に渡す。
    """
    z = compute_zscore(spread, cfg.lookback)
    result = BacktestResult(config=cfg)

    position = 0
    current: Trade | None = None

    for i in range(len(spread)):
        zi = z.iloc[i]
        if not np.isfinite(zi):
            continue

        si = float(spread.iloc[i])

        if position == 0:
            if zi > cfg.entry_z:
                # スプレッドが高い → 縮小に賭ける（スプレッド売り）
                position = -1
                current = Trade(i, spread.index[i], -1, float(zi), si)
            elif zi < -cfg.entry_z:
                position = +1
                current = Trade(i, spread.index[i], +1, float(zi), si)
            continue

        # 保有中
        assert current is not None
        holding = i - current.entry_index

        reason = ""
        if abs(zi) < cfg.exit_z:
            reason = "利確(z回帰)"
        elif abs(zi) > cfg.stop_z:
            reason = "損切り(z拡大)"
        elif holding >= cfg.max_holding_bars:
            reason = "時間切れ"

        if reason:
            current.exit_index = i
            current.exit_time = spread.index[i]
            current.exit_z = float(zi)
            current.exit_spread = si
            current.exit_reason = reason
            result.trades.append(current)
            position = 0
            current = None

    return result


def optimize_thresholds(spread: pd.Series, half_life_bars: float,
                        entry_grid: tuple = (1.5, 2.0, 2.5, 3.0),
                        exit_grid: tuple = (0.0, 0.5, 1.0),
                        fee_bps: float = 5.5) -> pd.DataFrame:
    """閾値の総当たり。ただし結果を鵜呑みにしないこと。

    ローリング窓はハーフライフから決める（恣意性を減らす）。
    """
    lookback = max(30, int(half_life_bars * 4))
    rows = []

    for entry in entry_grid:
        for exit_z in exit_grid:
            if exit_z >= entry:
                continue
            cfg = ZConfig(lookback=lookback, entry_z=entry,
                          exit_z=exit_z, stop_z=entry * 2,
                          fee_bps_per_leg=fee_bps)
            result = backtest_zscore(spread, cfg)
            s = result.summary()
            if s["取引数"] == 0:
                continue
            rows.append({
                "entry_z": entry,
                "exit_z": exit_z,
                "取引数": s["取引数"],
                "純利合計": s["純利合計"],
                "勝率": s["勝率"],
                "平均保有": s["平均保有バー"],
                "最大DD": s["最大DD"],
            })

    df = pd.DataFrame(rows)
    return df.sort_values("純利合計", ascending=False) if not df.empty else df


if __name__ == "__main__":
    from hedge_ratio import kalman_hedge
    from ou_process import estimate_ou

    prices = pd.read_parquet("./pair_prices.parquet")
    symbols = list(prices.columns)
    symbol_a, symbol_b = symbols[0], symbols[1]

    pair = np.log(prices[[symbol_a, symbol_b]].dropna())
    estimate = kalman_hedge(pair[symbol_a], pair[symbol_b])
    spread = estimate.spread

    params = estimate_ou(spread, bars_per_day=24.0)
    print(f"=== {symbol_a} / {symbol_b} ===")
    print(params.report())

    if not params.is_mean_reverting:
        print("\n平均回帰しないため、この戦略は適用できません。")
        raise SystemExit(0)

    lookback = max(30, int(params.half_life_bars * 4))
    print(f"\n  zスコアの窓: ハーフライフ×4 = {lookback}バー")

    cfg = ZConfig(lookback=lookback, entry_z=2.0, exit_z=0.5,
                  stop_z=4.0)
    result = backtest_zscore(spread, cfg)

    print(f"\n=== バックテスト (entry=±2.0 / exit=0.5) ===")
    for k, v in result.summary().items():
        if isinstance(v, dict):
            print(f"  {k:<14}: " + ", ".join(
                f"{kk} {vv}" for kk, vv in v.items()))
        elif isinstance(v, float):
            print(f"  {k:<14}: {v:>10.5f}")
        else:
            print(f"  {k:<14}: {v}")

    print("\n=== 閾値の総当たり ===")
    grid = optimize_thresholds(spread, params.half_life_bars)
    if not grid.empty:
        print(grid.head(10).to_string(
            index=False, float_format=lambda v: f"{v:.4f}"))

        print("\n!! 警告 !!")
        print(f"  {len(grid)}通りを試して最良を選ぶのは過剰最適化。")
        print("  記事32の Deflated Sharpe Ratio と CPCV で、")
        print("  その優位性が偶然でないことを確認すること。")
        print("  『最良のパラメータ』ではなく")
        print("  『広い範囲でプラスになる領域の中心』を選ぶのが実務。")

"""multiple_testing.py — 多重検定補正で偽ペアを排除する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

参考: Benjamini, Y. & Hochberg, Y. (1995)
        "Controlling the False Discovery Rate", JRSS-B 57(1)
      Bailey, D. & López de Prado, M. (2014)
        "The Deflated Sharpe Ratio"

  問題:
    N組を有意水準αで検定すると、
    真に共和分が無くても N×α 組が「有意」になる。
    N=10,000, α=0.05 なら 500組の偽陽性。

  3つの対処:
    1. Bonferroni : α を N で割る。厳格すぎて何も残らないことが多い
    2. FDR (BH法) : 「有意と判定したうちの偽陽性の割合」を制御。実用的
    3. 分割検定   : 前半で候補を選び、後半で確認する。最も直感的で強力

  実務では 2 と 3 を併用する。
"""
from __future__ import annotations

from dataclasses import dataclass, field

import numpy as np
import pandas as pd


@dataclass
class CorrectionResult:
    method: str
    n_tested: int
    n_significant_raw: int
    n_significant_corrected: int
    threshold: float
    expected_false_positives: float
    survivors: list = field(default_factory=list)

    def report(self) -> str:
        return "\n".join([
            f"[{self.method}]",
            f"  検定数            : {self.n_tested:,}",
            f"  補正前の有意       : {self.n_significant_raw:,}組",
            f"  期待される偽陽性   : {self.expected_false_positives:.1f}組",
            f"  補正後の閾値       : p < {self.threshold:.3e}",
            f"  補正後の有意       : {self.n_significant_corrected:,}組",
        ])


def bonferroni(results: list, alpha: float = 0.05) -> CorrectionResult:
    """Bonferroni補正。閾値を α/N にする。

    最も保守的。偽陽性はほぼ確実に防げるが、
    真のペアも一緒に落としてしまう（検出力が低い）。
    """
    n = len(results)
    if n == 0:
        return CorrectionResult("Bonferroni", 0, 0, 0, alpha, 0.0)

    threshold = alpha / n
    raw = [r for r in results if r.eg_pvalue < alpha]
    survivors = [r for r in results if r.eg_pvalue < threshold]

    return CorrectionResult(
        "Bonferroni", n, len(raw), len(survivors), threshold,
        n * alpha, survivors)


def benjamini_hochberg(results: list,
                       fdr: float = 0.10) -> CorrectionResult:
    """Benjamini-Hochberg法（FDR制御）。

    「有意と判定したもののうち、偽陽性である割合」を fdr 以下に抑える。
    Bonferroniより検出力が高く、実務で広く使われる。

    手順:
      1. p値を昇順に並べる
      2. p_(i) <= (i/N)·fdr を満たす最大の i を探す
      3. 上位 i 個を有意とする
    """
    n = len(results)
    if n == 0:
        return CorrectionResult("Benjamini-Hochberg", 0, 0, 0, fdr, 0.0)

    ordered = sorted(results, key=lambda r: r.eg_pvalue)
    pvalues = np.array([r.eg_pvalue for r in ordered])

    ranks = np.arange(1, n + 1)
    thresholds = ranks / n * fdr
    passing = pvalues <= thresholds

    if not passing.any():
        return CorrectionResult(
            "Benjamini-Hochberg", n,
            sum(1 for r in results if r.eg_pvalue < 0.05),
            0, 0.0, n * 0.05)

    k = int(np.max(np.where(passing)[0]))
    survivors = ordered[:k + 1]
    threshold = float(pvalues[k])

    return CorrectionResult(
        "Benjamini-Hochberg", n,
        sum(1 for r in results if r.eg_pvalue < 0.05),
        len(survivors), threshold, n * 0.05, survivors)


def split_sample_validation(prices: pd.DataFrame, results: list,
                            split_ratio: float = 0.6,
                            alpha: float = 0.05) -> CorrectionResult:
    """分割検定。前半で選び、後半で確認する。

    多重検定補正より直感的で、実質的に最も強力。
    「前半で有意だったペアが、後半でも有意か」を見るだけ。
    偶然のペアは後半で必ず落ちる。
    """
    from cointegration import test_pair

    split = int(len(prices) * split_ratio)
    first_half = prices.iloc[:split]
    second_half = prices.iloc[split:]

    print(f"  前半 {len(first_half)}本 / 後半 {len(second_half)}本")

    # 前半で有意だったペアだけを後半で再検定する
    in_sample = [r for r in results if r.eg_pvalue < alpha]
    print(f"  前半で有意: {len(in_sample)}組 → 後半で再検定")

    survivors = []
    for r in in_sample:
        try:
            retest = test_pair(second_half, r.symbol_a, r.symbol_b)
            if retest.eg_pvalue < alpha:
                survivors.append(retest)
        except Exception:
            continue

    survival_rate = len(survivors) / len(in_sample) if in_sample else 0.0
    print(f"  後半でも有意: {len(survivors)}組 (生存率 {survival_rate:.1%})")

    return CorrectionResult(
        "分割検定", len(results), len(in_sample), len(survivors),
        alpha, len(results) * alpha * alpha, survivors)


def stability_score(prices: pd.DataFrame, symbol_a: str, symbol_b: str,
                    n_folds: int = 5, alpha: float = 0.05) -> float:
    """期間を分割し、何割の期間で共和分が成立するかを測る。

    「たまたま全期間で有意」なペアと
    「どの期間でも安定して有意」なペアを区別する。
    """
    from cointegration import test_pair

    fold_size = len(prices) // n_folds
    if fold_size < 100:
        return 0.0

    significant = 0
    for i in range(n_folds):
        chunk = prices.iloc[i * fold_size:(i + 1) * fold_size]
        try:
            result = test_pair(chunk, symbol_a, symbol_b)
            if result.eg_pvalue < alpha:
                significant += 1
        except Exception:
            continue

    return significant / n_folds


if __name__ == "__main__":
    from cointegration import test_all
    from pair_universe import UniverseConfig, screen_pairs

    prices = pd.read_parquet("./pair_prices.parquet")
    candidates = screen_pairs(prices, UniverseConfig())

    print(f"\n=== 共和分検定 ({len(candidates)}組） ===")
    results = test_all(prices, candidates, verbose=False)

    print("\n" + "=" * 60)
    print(bonferroni(results).report())

    print()
    bh = benjamini_hochberg(results, fdr=0.10)
    print(bh.report())

    print("\n" + "=" * 60)
    print("[分割検定]")
    split = split_sample_validation(prices, results)

    print("\n" + "=" * 60)
    print("=== 最終的に残ったペア ===")

    final = split.survivors
    if not final:
        print("  ありません。")
        print("\n  これは失敗ではなく、正しい結論です。")
        print("  『安定した共和分ペアが見つからなかった』ということです。")
        print("  無理に取引すべきではありません。")
    else:
        print(f"\n  {len(final)}組が生き残りました。")
        print("  さらに期間安定性を確認します:\n")
        for r in sorted(final, key=lambda x: x.eg_pvalue)[:10]:
            score = stability_score(prices, r.symbol_a, r.symbol_b)
            mark = "  ← 安定" if score >= 0.8 else ""
            print(f"  {r.key:<26}p={r.eg_pvalue:.5f}  "
                  f"安定性={score:.0%}{mark}")

        print("\n  安定性80%以上のペアだけを実運用の候補にしてください。")

    print("\n※ 『何も残らなかった』を受け入れられるかが、")
    print("  この戦略で生き残れるかの分かれ目。")
    print("  無理に候補を作れば、それは過剰最適化そのもの（記事32）。")

"""pair_execution.py — 2銘柄を同時に建て、レッグリスクを管理する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  レッグリスク:
    A を買って B を売る間に価格が動くと、
    その瞬間は「A のロング」という方向性の賭けになっている。

  対処:
    1. バッチ注文（記事16）で1リクエストにまとめる
       → 完全な同時ではないが、往復が1回になりズレが最小化される
    2. 流動性の低い側を先に建てる
       → 約定しにくいほうを先に片付ける
    3. 片側が失敗したら、成功した側を即座に反対売買で解消する
       → 中途半端な状態を残さない
    4. 建玉額を揃える（数量ではなく金額）
       → β倍した数量が最小ロット未満にならないか事前確認（記事03）
"""
from __future__ import annotations

import time
from dataclasses import dataclass, field
from decimal import Decimal

from client import make_client
from symbol_rules import fetch_rules


@dataclass(frozen=True)
class LegSpec:
    symbol: str
    side: str
    qty: Decimal
    notional: Decimal

    def opposite_side(self) -> str:
        return "Sell" if self.side == "Buy" else "Buy"


@dataclass
class LegResult:
    spec: LegSpec
    ok: bool = False
    order_id: str = ""
    filled_qty: Decimal = Decimal(0)
    avg_price: Decimal = Decimal(0)
    ret_code: int = 0
    ret_msg: str = ""

    def line(self) -> str:
        if self.ok:
            return (f"  OK {self.spec.symbol:<12}{self.spec.side:<5}"
                    f"{self.filled_qty:>12} @{self.avg_price}")
        return (f"  NG {self.spec.symbol:<12}{self.spec.side:<5}"
                f"  {self.ret_code} {self.ret_msg}")


@dataclass
class PairExecution:
    leg_a: LegResult
    leg_b: LegResult
    unwound: bool = False
    elapsed_sec: float = 0.0

    @property
    def complete(self) -> bool:
        return self.leg_a.ok and self.leg_b.ok

    @property
    def has_leg_risk(self) -> bool:
        return self.leg_a.ok != self.leg_b.ok

    def report(self) -> str:
        lines = [f"=== ペア執行 ({self.elapsed_sec * 1000:.0f}ms) ==="]
        lines.append(self.leg_a.line())
        lines.append(self.leg_b.line())
        if self.complete:
            lines.append("  → 両レッグ成立。マーケットニュートラル")
        elif self.unwound:
            lines.append("  → 片側のみ成立したため解消しました")
        elif self.has_leg_risk:
            lines.append("  !! 片側のみ成立。方向性リスクを抱えています !!")
        else:
            lines.append("  → 両レッグ失敗。ポジションなし")
        return "\n".join(lines)


def size_legs(http, symbol_a: str, symbol_b: str, beta: float,
              notional_usd: Decimal, direction: int,
              category: str = "linear") -> tuple:
    """建玉額を揃えた2レッグを作る。

    direction: +1 なら「スプレッド買い」= A買い / B売り
               -1 なら「スプレッド売り」= A売り / B買い

    β は対数価格の回帰係数なので、
    建玉額の比を β にするのが理屈に合う。
    """
    r = http.get_tickers(category=category, symbol=symbol_a)
    price_a = Decimal(r["result"]["list"][0]["lastPrice"])
    r = http.get_tickers(category=category, symbol=symbol_b)
    price_b = Decimal(r["result"]["list"][0]["lastPrice"])

    rules_a = fetch_rules(http, symbol_a, category)
    rules_b = fetch_rules(http, symbol_b, category)

    beta_dec = Decimal(str(abs(beta)))
    # A の建玉額を基準に、B は β 倍
    notional_a = notional_usd
    notional_b = notional_usd * beta_dec

    qty_a = rules_a.round_qty(notional_a / price_a)
    qty_b = rules_b.round_qty(notional_b / price_b)

    if qty_a < rules_a.min_qty:
        raise ValueError(
            f"{symbol_a} の数量 {qty_a} が最小ロット {rules_a.min_qty} 未満です。"
            "建玉額を増やしてください")
    if qty_b < rules_b.min_qty:
        raise ValueError(
            f"{symbol_b} の数量 {qty_b} が最小ロット {rules_b.min_qty} 未満です。"
            f"β={beta:.4f} が小さすぎるか、建玉額が不足しています")

    side_a = "Buy" if direction > 0 else "Sell"
    side_b = "Sell" if direction > 0 else "Buy"

    return (LegSpec(symbol_a, side_a, qty_a, qty_a * price_a),
            LegSpec(symbol_b, side_b, qty_b, qty_b * price_b))


def execute_pair(http, leg_a: LegSpec, leg_b: LegSpec, *,
                 category: str = "linear",
                 use_batch: bool = True,
                 unwind_on_partial: bool = True) -> PairExecution:
    """2レッグを同時に建てる。"""
    started = time.perf_counter()
    stamp = int(time.time() * 1000) % 1_000_000

    result_a = LegResult(leg_a)
    result_b = LegResult(leg_b)

    if use_batch:
        # 1リクエストにまとめる（記事16）
        payload = []
        for i, leg in enumerate((leg_a, leg_b)):
            rules = fetch_rules(http, leg.symbol, category)
            payload.append({
                "symbol": leg.symbol,
                "side": leg.side,
                "orderType": "Market",
                "qty": rules.format_qty(leg.qty),
                "orderLinkId": f"pair-{stamp}-{i}",
            })

        r = http.place_batch_order(category=category, request=payload)
        ext = r.get("retExtInfo", {}).get("list", [])
        items = r["result"].get("list", []) if r["retCode"] == 0 else []

        for i, result in enumerate((result_a, result_b)):
            if r["retCode"] != 0:
                result.ret_code = r["retCode"]
                result.ret_msg = r["retMsg"]
                continue
            info = ext[i] if i < len(ext) else {"code": 0, "msg": ""}
            item = items[i] if i < len(items) else {}
            code = int(info.get("code", 0))
            result.ret_code = code
            result.ret_msg = info.get("msg", "")
            result.order_id = item.get("orderId", "")
            result.ok = code == 0 and bool(result.order_id)
    else:
        # 逐次発注。流動性の低い側を先に建てる
        for leg, result in ((leg_a, result_a), (leg_b, result_b)):
            rules = fetch_rules(http, leg.symbol, category)
            r = http.place_order(
                category=category, symbol=leg.symbol, side=leg.side,
                orderType="Market", qty=rules.format_qty(leg.qty),
                orderLinkId=f"pair-{stamp}-{leg.symbol[:6]}")
            result.ret_code = r["retCode"]
            result.ret_msg = r["retMsg"]
            if r["retCode"] == 0:
                result.ok = True
                result.order_id = r["result"]["orderId"]

    # 約定結果を取り込む
    time.sleep(0.4)
    for result in (result_a, result_b):
        if not result.ok or not result.order_id:
            continue
        r = http.get_order_history(category=category,
                                   symbol=result.spec.symbol,
                                   orderId=result.order_id, limit=1)
        if r["retCode"] == 0 and r["result"]["list"]:
            o = r["result"]["list"][0]
            result.filled_qty = Decimal(o.get("cumExecQty") or "0")
            result.avg_price = Decimal(o.get("avgPrice") or "0")

    execution = PairExecution(result_a, result_b,
                              elapsed_sec=time.perf_counter() - started)

    # 片側だけ成立した場合の解消
    if execution.has_leg_risk and unwind_on_partial:
        succeeded = result_a if result_a.ok else result_b
        print(f"  [レッグリスク] {succeeded.spec.symbol} のみ成立。解消します")
        rules = fetch_rules(http, succeeded.spec.symbol, category)
        r = http.place_order(
            category=category, symbol=succeeded.spec.symbol,
            side=succeeded.spec.opposite_side(), orderType="Market",
            qty=rules.format_qty(succeeded.filled_qty),
            reduceOnly=True,
            orderLinkId=f"unwind-{stamp}")
        execution.unwound = r["retCode"] == 0
        if not execution.unwound:
            print(f"  !! 解消に失敗: {r['retCode']} {r['retMsg']}")
            print("     手動で確認してください")

    return execution


def close_pair(http, symbol_a: str, symbol_b: str,
               category: str = "linear") -> PairExecution:
    """ペアポジションを両方とも決済する。"""
    r = http.get_positions(category=category, settleCoin="USDT")
    if r["retCode"] != 0:
        raise RuntimeError(f"get_positions失敗: {r['retCode']} {r['retMsg']}")

    positions = {p["symbol"]: p for p in r["result"]["list"]
                 if Decimal(p["size"] or "0") > 0}

    legs = []
    for symbol in (symbol_a, symbol_b):
        p = positions.get(symbol)
        if p is None:
            continue
        size = Decimal(p["size"])
        close_side = "Sell" if p["side"] == "Buy" else "Buy"
        legs.append(LegSpec(symbol, close_side, size,
                            size * Decimal(p.get("markPrice") or "0")))

    if len(legs) != 2:
        print(f"  [警告] ペアの片側しかポジションがありません "
              f"({len(legs)}件)")

    if not legs:
        raise RuntimeError("決済対象のポジションがありません")

    if len(legs) == 1:
        legs.append(LegSpec(symbol_b, "Buy", Decimal(0), Decimal(0)))

    return execute_pair(http, legs[0], legs[1], category=category,
                        unwind_on_partial=False)


if __name__ == "__main__":
    c = make_client()

    symbol_a, symbol_b = "ETHUSDT", "SOLUSDT"
    beta = 0.85

    print(f"=== ペア執行のテスト ({symbol_a} / {symbol_b}, β={beta}) ===\n")

    try:
        leg_a, leg_b = size_legs(c.http, symbol_a, symbol_b, beta,
                                 Decimal("200"), direction=+1)
        print(f"  レッグA: {leg_a.symbol} {leg_a.side} {leg_a.qty} "
              f"(建玉 {leg_a.notional:.2f} USD)")
        print(f"  レッグB: {leg_b.symbol} {leg_b.side} {leg_b.qty} "
              f"(建玉 {leg_b.notional:.2f} USD)")
        print(f"  建玉比: {leg_b.notional / leg_a.notional:.4f} "
              f"(目標 β={beta})\n")

        execution = execute_pair(c.http, leg_a, leg_b)
        print(execution.report())

        if execution.complete:
            time.sleep(2)
            print("\n=== 決済 ===")
            print(close_pair(c.http, symbol_a, symbol_b).report())

    except ValueError as e:
        print(f"  サイジングできません: {e}")

    print("\n※ バッチ発注でも完全な同時ではない。")
    print("  取引所内部の処理順序は保証されない。")
    print("  レッグリスクをゼロにはできないので、")
    print("  『片側だけ成立したら即座に解消する』設計が必須。")

"""pair_bot.py — ペアトレードbot（監視 → 発注 → 決済 → 崩壊時停止）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  構成:
    1. 起動時に対象ペアの共和分を再確認する（崩れていたら動かない）
    2. 定期的にスプレッドとzスコアを更新する
    3. 閾値を超えたらペア執行（サンプル7）
    4. 保有中はローリング検定で共和分の崩壊を監視する
    5. 崩壊を検知したら即座に決済して停止する

  記事24のサーキットブレーカーと組み合わせて使う前提。
  このbot自体はエントリー・エグジットのロジックだけを担当する。
"""
from __future__ import annotations

import json
import time
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from pathlib import Path

import numpy as np
import pandas as pd

from client import make_client
from cointegration import test_pair
from hedge_ratio import kalman_hedge
from kline_fetch import fetch_klines
from ou_process import estimate_ou
from pair_execution import close_pair, execute_pair, size_legs
from zscore_strategy import compute_zscore

STATE_PATH = Path("./pair_bot_state.json")


@dataclass(frozen=True)
class PairBotConfig:
    symbol_a: str
    symbol_b: str
    category: str = "linear"
    interval: str = "60"
    history_days: int = 90

    entry_z: float = 2.0
    exit_z: float = 0.5
    stop_z: float = 4.0
    max_holding_bars: int = 24 * 10

    notional_usd: Decimal = Decimal("200")
    poll_sec: int = 60

    # 共和分の崩壊判定
    coint_check_interval_sec: float = 3600.0
    coint_pvalue_threshold: float = 0.10
    coint_failures_to_stop: int = 3
    min_half_life_bars: float = 2.0
    max_half_life_bars: float = 24 * 14


@dataclass
class PairState:
    position: int = 0            # 0 / +1(スプレッド買い) / -1
    entry_time: float = 0.0
    entry_bar_index: int = 0
    entry_z: float = 0.0
    entry_beta: float = 0.0
    coint_failures: int = 0
    halted: bool = False
    halt_reason: str = ""

    def save(self, path: Path = STATE_PATH) -> None:
        path.write_text(json.dumps(asdict(self), ensure_ascii=False,
                                   indent=2), encoding="utf-8")

    @classmethod
    def load(cls, path: Path = STATE_PATH) -> "PairState":
        if not path.exists():
            return cls()
        try:
            return cls(**json.loads(path.read_text(encoding="utf-8")))
        except Exception:
            return cls()


class PairBot:
    def __init__(self, http, cfg: PairBotConfig):
        self.http = http
        self.cfg = cfg
        self.state = PairState.load()
        self.last_coint_check = 0.0
        self.bars_held = 0

    # --- データ ----------------------------------------------------

    def fetch_prices(self) -> pd.DataFrame:
        start = (datetime.now(timezone.utc)
                 - timedelta(days=self.cfg.history_days)).strftime("%Y-%m-%d")
        series = {}
        for symbol in (self.cfg.symbol_a, self.cfg.symbol_b):
            bars = fetch_klines(self.http, symbol,
                                interval=self.cfg.interval, start=start,
                                category=self.cfg.category, verbose=False)
            series[symbol] = bars["close"]
        return pd.DataFrame(series).dropna()

    def compute_signal(self, prices: pd.DataFrame) -> dict:
        """現在のスプレッド・zスコア・OUパラメータを計算する。"""
        pair = np.log(prices)
        estimate = kalman_hedge(pair[self.cfg.symbol_a],
                                pair[self.cfg.symbol_b])
        spread = estimate.spread

        params = estimate_ou(spread, bars_per_day=24.0)
        lookback = max(30, int(params.half_life_bars * 4)) \
            if params.is_mean_reverting else 24 * 5

        z = compute_zscore(spread, lookback)

        return {
            "spread": float(spread.iloc[-1]),
            "z": float(z.iloc[-1]) if np.isfinite(z.iloc[-1]) else 0.0,
            "beta": estimate.beta,
            "half_life": params.half_life_bars,
            "is_mean_reverting": params.is_mean_reverting,
            "lookback": lookback,
        }

    # --- 共和分の監視 ----------------------------------------------

    def check_cointegration(self, prices: pd.DataFrame,
                            signal: dict) -> tuple:
        """共和分が維持されているかを確認する。

        戻り値: (健全か, 理由)
        """
        result = test_pair(prices, self.cfg.symbol_a, self.cfg.symbol_b)

        if result.error:
            return False, f"検定に失敗: {result.error}"

        if result.eg_pvalue > self.cfg.coint_pvalue_threshold:
            return False, (f"共和分のp値が {result.eg_pvalue:.4f} に悪化 "
                           f"(閾値 {self.cfg.coint_pvalue_threshold})")

        if not signal["is_mean_reverting"]:
            return False, "スプレッドが平均回帰しなくなりました"

        hl = signal["half_life"]
        if hl < self.cfg.min_half_life_bars:
            return False, f"ハーフライフが短すぎます ({hl:.1f}バー)"
        if hl > self.cfg.max_half_life_bars:
            return False, f"ハーフライフが長すぎます ({hl:.1f}バー)"

        return True, f"共和分は維持されています (p={result.eg_pvalue:.4f})"

    # --- 売買 ------------------------------------------------------

    def enter(self, direction: int, signal: dict) -> bool:
        try:
            leg_a, leg_b = size_legs(
                self.http, self.cfg.symbol_a, self.cfg.symbol_b,
                signal["beta"], self.cfg.notional_usd, direction,
                self.cfg.category)
        except ValueError as e:
            print(f"  サイジング失敗: {e}")
            return False

        execution = execute_pair(self.http, leg_a, leg_b,
                                 category=self.cfg.category)
        print(execution.report())

        if not execution.complete:
            return False

        self.state.position = direction
        self.state.entry_time = time.time()
        self.state.entry_z = signal["z"]
        self.state.entry_beta = signal["beta"]
        self.bars_held = 0
        self.state.save()
        return True

    def exit(self, reason: str) -> bool:
        print(f"  決済: {reason}")
        try:
            execution = close_pair(self.http, self.cfg.symbol_a,
                                   self.cfg.symbol_b, self.cfg.category)
            print(execution.report())
        except RuntimeError as e:
            print(f"  決済失敗: {e}")
            return False

        self.state.position = 0
        self.state.entry_time = 0.0
        self.bars_held = 0
        self.state.save()
        return True

    def halt(self, reason: str) -> None:
        self.state.halted = True
        self.state.halt_reason = reason
        self.state.save()
        print(f"\n!! 停止: {reason} !!")
        print("  再開するには pair_bot_state.json を削除してください。")

    # --- メインループ ----------------------------------------------

    def step(self) -> None:
        prices = self.fetch_prices()
        if len(prices) < 200:
            print("  データ不足。スキップします")
            return

        signal = self.compute_signal(prices)
        z = signal["z"]

        print(f"  z={z:>+7.3f}  β={signal['beta']:>8.4f}  "
              f"HL={signal['half_life']:>6.1f}バー  "
              f"pos={self.state.position:+d}")

        # 共和分の定期確認
        now = time.time()
        if now - self.last_coint_check > self.cfg.coint_check_interval_sec:
            self.last_coint_check = now
            healthy, reason = self.check_cointegration(prices, signal)

            if healthy:
                self.state.coint_failures = 0
            else:
                self.state.coint_failures += 1
                print(f"  [共和分の警告 "
                      f"{self.state.coint_failures}/"
                      f"{self.cfg.coint_failures_to_stop}] {reason}")

                if self.state.coint_failures >= self.cfg.coint_failures_to_stop:
                    if self.state.position != 0:
                        self.exit("共和分の崩壊")
                    self.halt(f"共和分の崩壊: {reason}")
                    return

        # ポジションなし → エントリー判定
        if self.state.position == 0:
            if not signal["is_mean_reverting"]:
                return
            if z > self.cfg.entry_z:
                print(f"  エントリー: スプレッド売り (z={z:.3f})")
                self.enter(-1, signal)
            elif z < -self.cfg.entry_z:
                print(f"  エントリー: スプレッド買い (z={z:.3f})")
                self.enter(+1, signal)
            return

        # 保有中 → エグジット判定
        self.bars_held += 1

        if abs(z) < self.cfg.exit_z:
            self.exit(f"利確 (z={z:.3f} が {self.cfg.exit_z} 以内に回帰)")
        elif abs(z) > self.cfg.stop_z:
            self.exit(f"損切り (z={z:.3f} が {self.cfg.stop_z} を超過)")
        elif self.bars_held >= self.cfg.max_holding_bars:
            self.exit(f"時間切れ ({self.bars_held}バー保有)")

    def run(self) -> None:
        print(f"=== ペアトレードbot "
              f"({self.cfg.symbol_a} / {self.cfg.symbol_b}) ===")

        if self.state.halted:
            print(f"  停止中: {self.state.halt_reason}")
            print("  pair_bot_state.json を削除すると再開できます。")
            return

        # 起動時の健全性確認
        prices = self.fetch_prices()
        signal = self.compute_signal(prices)
        healthy, reason = self.check_cointegration(prices, signal)
        print(f"  起動時の確認: {reason}")

        if not healthy:
            self.halt(f"起動時に共和分が成立していません: {reason}")
            return

        print(f"  entry=±{self.cfg.entry_z} / exit={self.cfg.exit_z} / "
              f"stop=±{self.cfg.stop_z}")
        print("  Ctrl+C で停止\n")

        try:
            while not self.state.halted:
                try:
                    self.step()
                except Exception as e:
                    print(f"  [error] {type(e).__name__}: {e}")
                time.sleep(self.cfg.poll_sec)
        except KeyboardInterrupt:
            print("\n停止します。")
            if self.state.position != 0:
                answer = input("  ポジションを決済しますか? [y/N] ")
                if answer.lower() == "y":
                    self.exit("手動停止")


if __name__ == "__main__":
    c = make_client()

    cfg = PairBotConfig(
        symbol_a="ETHUSDT", symbol_b="SOLUSDT",
        entry_z=2.0, exit_z=0.5, stop_z=4.0,
        notional_usd=Decimal("200"),
        poll_sec=60)

    PairBot(c.http, cfg).run()

    # 実運用の前に:
    #   1. サンプル6の多重検定補正でペアを選ぶ
    #   2. 記事32の CPCV と Deflated Sharpe Ratio で検証する
    #   3. testnet で数週間動かす
    #   4. 記事24のサーキットブレーカーを組み込む
    #   5. 最小ロットで開始する

"""test_pairs.py — ペアトレードのロジックを検証する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pytest test_pairs.py -v

  統計的な処理は「正しく見えるが間違っている」ことが起きやすい。
  既知の答えがある人工データで検証するのが確実。
"""
from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from hedge_ratio import kalman_hedge, ols_hedge, tls_hedge
from multiple_testing import benjamini_hochberg
from ou_process import estimate_ou, half_life_simple
from zscore_strategy import ZConfig, backtest_zscore, compute_zscore


def make_ou_series(theta: float = 0.05, mu: float = 0.0,
                   sigma: float = 0.01, n: int = 2000,
                   seed: int = 42) -> pd.Series:
    """既知のパラメータでOU過程を生成する。"""
    rng = np.random.default_rng(seed)
    x = np.zeros(n)
    for t in range(1, n):
        x[t] = x[t - 1] + theta * (mu - x[t - 1]) + sigma * rng.normal()
    return pd.Series(x, index=pd.date_range("2025-01-01", periods=n,
                                            freq="h"))


# --- OU過程 ---------------------------------------------------------

def test_ou_recovers_known_theta():
    """既知のθを推定で復元できること。"""
    true_theta = 0.05
    series = make_ou_series(theta=true_theta, n=5000)
    params = estimate_ou(series)
    assert params.theta == pytest.approx(true_theta, rel=0.15)


def test_ou_half_life_matches_formula():
    """ハーフライフ = ln(2)/θ であること。"""
    series = make_ou_series(theta=0.05, n=5000)
    params = estimate_ou(series)
    assert params.half_life_bars == pytest.approx(
        np.log(2) / params.theta, rel=1e-6)


def test_random_walk_is_not_mean_reverting():
    """ランダムウォークは平均回帰しないと判定されること。

    これを見逃すと、平均回帰しない系列で取引してしまう。
    """
    rng = np.random.default_rng(7)
    walk = pd.Series(np.cumsum(rng.normal(0, 0.01, 3000)))
    params = estimate_ou(walk)
    # θがほぼ0 → ハーフライフが非常に長いか無限
    assert params.half_life_bars > 500 or not params.is_mean_reverting


def test_half_life_simple_matches_full_estimate():
    series = make_ou_series(theta=0.03, n=3000)
    full = estimate_ou(series).half_life_bars
    simple = half_life_simple(series)
    assert simple == pytest.approx(full, rel=0.01)


# --- ヘッジ比率 -----------------------------------------------------

def make_cointegrated_pair(beta: float = 1.5, n: int = 2000,
                           seed: int = 1) -> tuple:
    """既知のβで共和分する2系列を作る。"""
    rng = np.random.default_rng(seed)
    b = np.cumsum(rng.normal(0, 0.01, n)) + 10
    spread = make_ou_series(theta=0.05, sigma=0.01, n=n, seed=seed + 1)
    a = beta * b + spread.to_numpy()
    index = pd.date_range("2025-01-01", periods=n, freq="h")
    return pd.Series(a, index=index), pd.Series(b, index=index)


def test_ols_recovers_known_beta():
    a, b = make_cointegrated_pair(beta=1.5)
    assert ols_hedge(a, b).beta == pytest.approx(1.5, rel=0.05)


def test_tls_recovers_known_beta():
    a, b = make_cointegrated_pair(beta=1.5)
    assert tls_hedge(a, b).beta == pytest.approx(1.5, rel=0.10)


def test_kalman_converges_to_true_beta():
    a, b = make_cointegrated_pair(beta=1.5, n=3000)
    estimate = kalman_hedge(a, b)
    assert estimate.beta == pytest.approx(1.5, rel=0.10)
    assert estimate.beta_series is not None
    assert len(estimate.beta_series) == len(a)


def test_ols_spread_is_stationary():
    """OLSの残差が平均回帰すること。"""
    a, b = make_cointegrated_pair(beta=1.5, n=3000)
    spread = ols_hedge(a, b).spread
    assert estimate_ou(spread).is_mean_reverting


# --- zスコア --------------------------------------------------------

def test_zscore_does_not_peek_future():
    """zスコアの計算で未来を参照していないこと。

    最重要のテスト。shift(1) を忘れると
    バックテストが劇的に良く見えるが、実運用では機能しない。
    """
    series = make_ou_series(n=500)
    z_full = compute_zscore(series, lookback=50)

    # 途中までのデータで計算しても、その時点の値は変わらないはず
    truncated = series.iloc[:300]
    z_truncated = compute_zscore(truncated, lookback=50)

    common = z_truncated.dropna().index
    for idx in common[-20:]:
        assert z_full.loc[idx] == pytest.approx(
            z_truncated.loc[idx], rel=1e-9)


def test_zscore_is_standardized():
    """zスコアの標準偏差が概ね1になること。"""
    series = make_ou_series(n=3000)
    z = compute_zscore(series, lookback=100).dropna()
    assert z.std() == pytest.approx(1.0, abs=0.3)


def test_backtest_enters_on_threshold():
    """閾値を超えたらエントリーすること。"""
    series = make_ou_series(theta=0.1, sigma=0.02, n=2000)
    cfg = ZConfig(lookback=50, entry_z=2.0, exit_z=0.5, stop_z=4.0,
                  fee_bps_per_leg=0.0)
    result = backtest_zscore(series, cfg)
    assert result.n_trades > 0
    for trade in result.trades:
        assert abs(trade.entry_z) >= cfg.entry_z


def test_fees_reduce_returns():
    """手数料を入れると純利が粗利より小さくなること。"""
    series = make_ou_series(theta=0.1, sigma=0.02, n=2000)
    base = ZConfig(lookback=50, fee_bps_per_leg=0.0)
    with_fee = ZConfig(lookback=50, fee_bps_per_leg=5.5)

    free = backtest_zscore(series, base)
    paid = backtest_zscore(series, with_fee)

    assert sum(paid.net_returns()) < sum(free.net_returns())


# --- 多重検定 -------------------------------------------------------

class FakeResult:
    def __init__(self, pvalue: float, i: int):
        self.eg_pvalue = pvalue
        self.symbol_a = f"A{i}"
        self.symbol_b = f"B{i}"
        self.key = f"A{i}/B{i}"


def test_bh_is_less_strict_than_bonferroni():
    """BH法はBonferroniより多くのペアを通すこと。"""
    from multiple_testing import bonferroni

    rng = np.random.default_rng(0)
    # 大半は帰無仮説が真（一様分布）、一部だけ本当に有意
    pvalues = list(rng.uniform(0, 1, 950)) + list(rng.uniform(0, 0.001, 50))
    results = [FakeResult(p, i) for i, p in enumerate(pvalues)]

    bonf = bonferroni(results)
    bh = benjamini_hochberg(results, fdr=0.10)

    assert bh.n_significant_corrected >= bonf.n_significant_corrected


def test_bh_rejects_pure_noise():
    """全てが帰無仮説（一様分布のp値）なら、ほとんど通らないこと。"""
    rng = np.random.default_rng(3)
    results = [FakeResult(p, i)
               for i, p in enumerate(rng.uniform(0, 1, 1000))]
    bh = benjamini_hochberg(results, fdr=0.10)
    # ノイズだけなら生存は極めて少ないはず
    assert bh.n_significant_corrected < 10