Bybit APIで約定履歴・実現損益・手数料を集計するPython実装
7日窓の壁を越える

"""exec_fetch.py — 約定履歴を7日窓ループで全期間取得する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

制約: endTime - startTime <= 7日 / データ保持は過去2年
"""
from __future__ import annotations

import time
from collections.abc import Iterator
from datetime import datetime, timedelta, timezone

import pandas as pd

WINDOW = timedelta(days=7)          # APIの上限
PAGE_LIMIT = 100                    # 1ページの上限
MAX_PAGES_PER_WINDOW = 500          # 安全上限

NUMERIC_COLS = ["execPrice", "execQty", "execFee", "execValue", "feeRate",
                "closedSize", "orderQty", "orderPrice", "leavesQty",
                "markPrice", "indexPrice", "execPnl"]


def windows(start: datetime, end: datetime,
            size: timedelta = WINDOW) -> Iterator[tuple[datetime, datetime]]:
    """[start, end) を size ごとの窓に分割する。"""
    cur = start
    while cur < end:
        nxt = min(cur + size, end)
        yield cur, nxt
        cur = nxt


def fetch_executions(http, start: str | datetime, end: str | datetime | None = None,
                     category: str = "linear", symbol: str | None = None,
                     sleep: float = 0.08, verbose: bool = True) -> pd.DataFrame:
    """全期間の約定履歴を取得する（7日窓ループ）。"""
    if isinstance(start, str):
        start = datetime.fromisoformat(start)
    if start.tzinfo is None:
        start = start.replace(tzinfo=timezone.utc)
    end = end or datetime.now(timezone.utc)
    if isinstance(end, str):
        end = datetime.fromisoformat(end)
    if end.tzinfo is None:
        end = end.replace(tzinfo=timezone.utc)

    all_rows: list[dict] = []
    win_list = list(windows(start, end))

    for i, (w_start, w_end) in enumerate(win_list, 1):
        cursor = None
        win_rows = 0

        for _ in range(MAX_PAGES_PER_WINDOW):
            kw: dict = {
                "category": category,
                "startTime": int(w_start.timestamp() * 1000),
                "endTime": int(w_end.timestamp() * 1000),
                "limit": PAGE_LIMIT,
            }
            if symbol:
                kw["symbol"] = symbol
            if cursor:
                kw["cursor"] = cursor

            res = http.get_executions(**kw)
            if res["retCode"] != 0:
                print(f"[warn] {w_start:%Y-%m-%d}: "
                      f"{res['retCode']} {res['retMsg']}")
                break

            page = res["result"]["list"]
            all_rows.extend(page)
            win_rows += len(page)

            cursor = res["result"].get("nextPageCursor")
            if not cursor or not page:
                break
            time.sleep(sleep)

        if verbose and (win_rows or i % 10 == 0):
            print(f"  [{i}/{len(win_list)}] {w_start:%Y-%m-%d} 〜 "
                  f"{w_end:%Y-%m-%d}: {win_rows}件 (累計 {len(all_rows)})")
        time.sleep(sleep)

    if not all_rows:
        return pd.DataFrame()

    df = pd.DataFrame(all_rows)
    df["ts"] = pd.to_datetime(df["execTime"].astype("int64"), unit="ms",
                              utc=True)
    for col in NUMERIC_COLS:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0.0)
    if "isMaker" in df.columns:
        df["isMaker"] = df["isMaker"].astype(bool)

    df = (df.drop_duplicates(subset="execId")
            .set_index("ts")
            .sort_index())

    if verbose:
        print(f"[完了] {len(df)}件  {df.index[0]:%Y-%m-%d} 〜 "
              f"{df.index[-1]:%Y-%m-%d}")
    return df


def demonstrate_window_trap(http, category: str = "linear") -> None:
    """startTimeに1年前を渡すと7日分しか返らないことを示す。"""
    one_year_ago = datetime.now(timezone.utc) - timedelta(days=365)

    res = http.get_executions(
        category=category,
        startTime=int(one_year_ago.timestamp() * 1000),
        limit=100,
    )
    rows = res["result"]["list"] if res["retCode"] == 0 else []
    if rows:
        times = [int(r["execTime"]) for r in rows]
        span_days = (max(times) - min(times)) / 86_400_000
        print(f"startTime=1年前 で取得 -> {len(rows)}件 / "
              f"期間 {span_days:.1f}日分")
        print("  -> 7日窓しか返らない。ループが必須。")
    else:
        print(f"取得なし (retCode={res['retCode']} {res['retMsg']})")


if __name__ == "__main__":
    from client import make_client

    c = make_client()

    print("=== 7日窓の罠を確認 ===")
    demonstrate_window_trap(c.http)

    print("\n=== 正しく90日分を取得 ===")
    start = datetime.now(timezone.utc) - timedelta(days=90)
    df = fetch_executions(c.http, start, category="linear")
    if not df.empty:
        print(f"\n約定 {len(df)}件 / 銘柄 {df['symbol'].nunique()}種")
        print(f"手数料合計: {df['execFee'].sum():,.4f}")

"""pnl_fetch.py — 決済損益（closed PnL）を7日窓ループで全期間取得する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

注意: closed_pnl は linear / inverse のみ。現物は非対応。
"""
from __future__ import annotations

import time
from datetime import datetime, timedelta, timezone

import pandas as pd

from exec_fetch import windows

NUMERIC_COLS = ["qty", "orderPrice", "closedSize", "cumEntryValue",
                "avgEntryPrice", "cumExitValue", "avgExitPrice",
                "closedPnl", "leverage", "openFee", "closeFee", "fillCount"]


def fetch_closed_pnl(http, start: str | datetime,
                     end: str | datetime | None = None,
                     category: str = "linear", symbol: str | None = None,
                     sleep: float = 0.08, verbose: bool = True) -> pd.DataFrame:
    if isinstance(start, str):
        start = datetime.fromisoformat(start)
    if start.tzinfo is None:
        start = start.replace(tzinfo=timezone.utc)
    end = end or datetime.now(timezone.utc)
    if isinstance(end, str):
        end = datetime.fromisoformat(end)
    if end.tzinfo is None:
        end = end.replace(tzinfo=timezone.utc)

    rows: list[dict] = []
    win_list = list(windows(start, end))

    for i, (w_start, w_end) in enumerate(win_list, 1):
        cursor = None
        for _ in range(500):
            kw: dict = {
                "category": category,
                "startTime": int(w_start.timestamp() * 1000),
                "endTime": int(w_end.timestamp() * 1000),
                "limit": 100,
            }
            if symbol:
                kw["symbol"] = symbol
            if cursor:
                kw["cursor"] = cursor

            res = http.get_closed_pnl(**kw)
            if res["retCode"] != 0:
                print(f"[warn] {w_start:%Y-%m-%d}: {res['retMsg']}")
                break

            page = res["result"]["list"]
            rows.extend(page)
            cursor = res["result"].get("nextPageCursor")
            if not cursor or not page:
                break
            time.sleep(sleep)

        if verbose and i % 10 == 0:
            print(f"  [{i}/{len(win_list)}] 累計 {len(rows)}件")
        time.sleep(sleep)

    if not rows:
        return pd.DataFrame()

    df = pd.DataFrame(rows)
    df["ts"] = pd.to_datetime(df["updatedTime"].astype("int64"), unit="ms",
                              utc=True)
    for col in NUMERIC_COLS:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0.0)

    df = df.drop_duplicates(subset=["orderId", "updatedTime"])
    df = df.set_index("ts").sort_index()

    # 手数料を含む正味の損益を作る（closedPnlは手数料込みの定義だが、
    # openFee/closeFeeも別に返るため、両方で検算できるようにしておく）
    df["net_pnl"] = df["closedPnl"]
    df["total_fee"] = df["openFee"] + df["closeFee"]
    df["gross_pnl"] = df["net_pnl"] + df["total_fee"]

    if verbose:
        print(f"[完了] 決済 {len(df)}件  {df.index[0]:%Y-%m-%d} 〜 "
              f"{df.index[-1]:%Y-%m-%d}")
    return df


def trade_stats(df: pd.DataFrame) -> dict:
    """トレード単位の基本統計。"""
    if df.empty:
        return {"error": "データなし"}

    pnl = df["net_pnl"]
    wins = pnl[pnl > 0]
    losses = pnl[pnl < 0]

    gross_win = float(wins.sum())
    gross_loss = float(-losses.sum())

    return {
        "トレード数": len(pnl),
        "勝ち": len(wins),
        "負け": len(losses),
        "勝率": f"{len(wins) / len(pnl):.1%}" if len(pnl) else "-",
        "総利益": round(gross_win, 4),
        "総損失": round(gross_loss, 4),
        "純損益": round(float(pnl.sum()), 4),
        "手数料合計": round(float(df["total_fee"].sum()), 4),
        "プロフィットファクター": (round(gross_win / gross_loss, 3)
                                 if gross_loss else float("inf")),
        "平均利益": round(float(wins.mean()), 4) if len(wins) else 0,
        "平均損失": round(float(losses.mean()), 4) if len(losses) else 0,
        "ペイオフレシオ": (round(abs(wins.mean() / losses.mean()), 3)
                          if len(wins) and len(losses) else "-"),
        "最大利益": round(float(pnl.max()), 4),
        "最大損失": round(float(pnl.min()), 4),
        "期待値": round(float(pnl.mean()), 4),
    }


if __name__ == "__main__":
    from client import make_client

    c = make_client()
    start = datetime.now(timezone.utc) - timedelta(days=90)
    df = fetch_closed_pnl(c.http, start)

    if df.empty:
        print("決済履歴がありません")
    else:
        print("\n=== トレード統計 ===")
        for k, v in trade_stats(df).items():
            print(f"  {k:<22}: {v}")

        print("\n=== 銘柄別 ===")
        by_sym = df.groupby("symbol").agg(
            n=("net_pnl", "count"), pnl=("net_pnl", "sum"),
            fee=("total_fee", "sum"),
        ).sort_values("pnl")
        print(by_sym.to_string(float_format=lambda v: f"{v:,.4f}"))

"""fee_analysis.py — 手数料をMaker/Takerに分解して分析する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import pandas as pd

# 通常の約定のみを対象にする（Funding / Settle 等を除く）
TRADE_EXEC_TYPES = {"Trade"}


def fee_breakdown(execs: pd.DataFrame) -> dict:
    """Maker/Taker別の手数料と実効コストを算出する。"""
    if execs.empty:
        return {"error": "データなし"}

    df = execs[execs["execType"].isin(TRADE_EXEC_TYPES)].copy()
    if df.empty:
        return {"error": "通常約定がありません"}

    maker = df[df["isMaker"]]
    taker = df[~df["isMaker"]]

    total_value = float(df["execValue"].sum())
    total_fee = float(df["execFee"].sum())

    def side_stats(sub: pd.DataFrame, label: str) -> dict:
        value = float(sub["execValue"].sum())
        fee = float(sub["execFee"].sum())
        return {
            f"{label}_約定数": len(sub),
            f"{label}_約定金額": round(value, 2),
            f"{label}_手数料": round(fee, 4),
            f"{label}_実効レート": (f"{fee / value * 100:.5f}%"
                                   if value else "-"),
            f"{label}_bps": round(fee / value * 10_000, 2) if value else 0,
        }

    out = {
        "総約定数": len(df),
        "総約定金額": round(total_value, 2),
        "総手数料": round(total_fee, 4),
        "平均実効レート": (f"{total_fee / total_value * 100:.5f}%"
                          if total_value else "-"),
        "Maker比率(件数)": f"{len(maker) / len(df):.1%}",
        "Maker比率(金額)": (f"{float(maker['execValue'].sum()) / total_value:.1%}"
                           if total_value else "-"),
    }
    out.update(side_stats(maker, "Maker"))
    out.update(side_stats(taker, "Taker"))

    # 全部メイカーにできたら、いくら節約できたか
    if len(maker) and len(taker):
        maker_rate = float(maker["execFee"].sum()) / float(maker["execValue"].sum())
        taker_value = float(taker["execValue"].sum())
        taker_fee = float(taker["execFee"].sum())
        savings = taker_fee - taker_value * maker_rate
        out["全Maker化で節約できる額"] = round(savings, 4)
        out["節約率"] = (f"{savings / total_fee:.1%}" if total_fee else "-")

    return out


def fee_by_symbol(execs: pd.DataFrame, top: int = 15) -> pd.DataFrame:
    df = execs[execs["execType"].isin(TRADE_EXEC_TYPES)].copy()
    agg = df.groupby("symbol").agg(
        n=("execFee", "count"),
        value=("execValue", "sum"),
        fee=("execFee", "sum"),
        maker_ratio=("isMaker", "mean"),
    )
    agg["fee_bps"] = agg["fee"] / agg["value"] * 10_000
    return agg.sort_values("fee", ascending=False).head(top)


def fee_timeline(execs: pd.DataFrame, freq: str = "W") -> pd.DataFrame:
    df = execs[execs["execType"].isin(TRADE_EXEC_TYPES)].copy()
    agg = df.resample(freq).agg(
        n=("execFee", "count"),
        value=("execValue", "sum"),
        fee=("execFee", "sum"),
        maker_ratio=("isMaker", "mean"),
    )
    agg["fee_bps"] = (agg["fee"] / agg["value"] * 10_000).fillna(0)
    return agg


def non_trade_summary(execs: pd.DataFrame) -> pd.DataFrame:
    """Trade以外のexecType（Funding / Settle / 清算など）を集計する。

    これらを通常の約定と混ぜて集計すると損益がおかしくなる。
    """
    df = execs[~execs["execType"].isin(TRADE_EXEC_TYPES)]
    if df.empty:
        return pd.DataFrame()
    return df.groupby("execType").agg(
        n=("execFee", "count"),
        fee=("execFee", "sum"),
        pnl=("execPnl", "sum") if "execPnl" in df.columns else ("execFee", "sum"),
    )


if __name__ == "__main__":
    from datetime import datetime, timedelta, timezone

    from client import make_client
    from exec_fetch import fetch_executions

    c = make_client()
    start = datetime.now(timezone.utc) - timedelta(days=60)
    execs = fetch_executions(c.http, start, category="linear", verbose=False)

    if execs.empty:
        raise SystemExit("約定履歴がありません")

    print("=== 手数料の内訳 ===")
    for k, v in fee_breakdown(execs).items():
        print(f"  {k:<26}: {v}")

    print("\n=== 銘柄別（手数料の多い順） ===")
    print(fee_by_symbol(execs).to_string(
        float_format=lambda v: f"{v:,.4f}"))

    print("\n=== 週次推移 ===")
    print(fee_timeline(execs).to_string(
        float_format=lambda v: f"{v:,.4f}"))

    nt = non_trade_summary(execs)
    if not nt.empty:
        print("\n=== Trade以外のexecType ===")
        print(nt.to_string(float_format=lambda v: f"{v:,.4f}"))
        print("  ※ これらは通常の約定と分けて集計する必要がある")

"""strategy_pnl.py — orderLinkIdのプレフィックスで戦略別に損益を分解する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

前提: 発注時に orderLinkId = "戦略名-..." の形で付けていること（記事17）。
"""
from __future__ import annotations

import re

import pandas as pd

UNKNOWN = "(不明)"


def extract_strategy(link_id: str, known: set[str] | None = None) -> str:
    """orderLinkIdから戦略名を取り出す。

    known を渡せば、既知の戦略名に前方一致するものだけを採用する。
    渡さない場合は最初のハイフンまでを戦略名とみなす。
    """
    if not link_id:
        return UNKNOWN
    if known:
        for k in sorted(known, key=len, reverse=True):
            if link_id.startswith(k):
                return k
        return UNKNOWN
    m = re.match(r"^([A-Za-z0-9_]+?)-", link_id)
    return m.group(1) if m else UNKNOWN


def attach_strategy(df: pd.DataFrame, known: set[str] | None = None,
                    col: str = "orderLinkId") -> pd.DataFrame:
    out = df.copy()
    if col not in out.columns:
        out["strategy"] = UNKNOWN
        return out
    out["strategy"] = out[col].fillna("").map(
        lambda v: extract_strategy(v, known))
    return out


def strategy_report(execs: pd.DataFrame, pnl: pd.DataFrame,
                    known: set[str] | None = None) -> pd.DataFrame:
    """戦略ごとの損益・手数料・執行品質をまとめる。"""
    e = attach_strategy(execs[execs["execType"] == "Trade"], known)
    p = attach_strategy(pnl, known) if not pnl.empty else pd.DataFrame()

    fee_agg = e.groupby("strategy").agg(
        約定数=("execFee", "count"),
        約定金額=("execValue", "sum"),
        手数料=("execFee", "sum"),
        Maker率=("isMaker", "mean"),
    )
    fee_agg["手数料bps"] = (fee_agg["手数料"] / fee_agg["約定金額"]
                           * 10_000).fillna(0)

    if p.empty:
        return fee_agg

    pnl_agg = p.groupby("strategy").agg(
        決済数=("net_pnl", "count"),
        純損益=("net_pnl", "sum"),
        勝ち=("net_pnl", lambda s: int((s > 0).sum())),
    )
    pnl_agg["勝率"] = (pnl_agg["勝ち"] / pnl_agg["決済数"]).fillna(0)
    pnl_agg["平均損益"] = pnl_agg["純損益"] / pnl_agg["決済数"]

    merged = fee_agg.join(pnl_agg, how="outer").fillna(0)
    # 手数料が損益をどれだけ食っているか
    merged["手数料/粗利"] = merged.apply(
        lambda r: (r["手数料"] / abs(r["純損益"] + r["手数料"])
                   if (r["純損益"] + r["手数料"]) else 0), axis=1)
    return merged.sort_values("純損益", ascending=False)


def symbol_by_strategy(execs: pd.DataFrame,
                       known: set[str] | None = None) -> pd.DataFrame:
    """戦略×銘柄のクロス集計。どの戦略がどの銘柄で稼いでいるか。"""
    e = attach_strategy(execs[execs["execType"] == "Trade"], known)
    return e.pivot_table(index="strategy", columns="symbol",
                         values="execValue", aggfunc="sum",
                         fill_value=0)


if __name__ == "__main__":
    from datetime import datetime, timedelta, timezone

    from client import make_client
    from exec_fetch import fetch_executions
    from pnl_fetch import fetch_closed_pnl

    c = make_client()
    start = datetime.now(timezone.utc) - timedelta(days=60)

    execs = fetch_executions(c.http, start, category="linear", verbose=False)
    pnl = fetch_closed_pnl(c.http, start, category="linear", verbose=False)

    if execs.empty:
        raise SystemExit("約定履歴がありません")

    # 自分が使っている戦略名を列挙しておくと分類が正確になる
    KNOWN = {"trendfol", "gridbot", "carry", "meanrev", "demo"}

    print("=== 戦略別サマリ ===")
    rep = strategy_report(execs, pnl, KNOWN)
    with pd.option_context("display.width", 160):
        print(rep.to_string(float_format=lambda v: f"{v:,.4f}"))

    print("\n--- 読み方 ---")
    print("  手数料/粗利 が 0.5 を超える戦略は、利益の半分を手数料で")
    print("  失っている。戦略の改善より執行の改善が優先（記事13）。")
    print("  Maker率が低い戦略は、PostOnly化の余地がある。")

"""exec_quality.py — 執行品質（スリッページ・実効スプレッド）を測る
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

指標:
  - 実効スプレッドコスト: 約定価格 vs 同時刻のマーク価格
  - 指値の乖離        : 注文価格 vs 約定価格（成行では発生する）
  - Implementation Shortfall の近似
"""
from __future__ import annotations

import pandas as pd


def slippage_vs_mark(execs: pd.DataFrame) -> pd.DataFrame:
    """約定価格とマーク価格の差をbpsで算出する。

    markPrice は約定時点の値がレスポンスに含まれるため、
    「その時点の公正価格に対してどれだけ不利に約定したか」が測れる。
    """
    df = execs[(execs["execType"] == "Trade")
               & (execs.get("markPrice", 0) > 0)].copy()
    if df.empty:
        return df

    # 買いは高く約定すると不利、売りは安く約定すると不利
    sign = df["side"].map({"Buy": 1, "Sell": -1}).fillna(0)
    df["slip_bps"] = ((df["execPrice"] - df["markPrice"])
                      / df["markPrice"] * 10_000 * sign)
    return df


def order_price_deviation(execs: pd.DataFrame) -> pd.DataFrame:
    """注文価格と約定価格の乖離。指値なら0付近、成行では乖離が出る。"""
    df = execs[(execs["execType"] == "Trade")
               & (execs.get("orderPrice", 0) > 0)].copy()
    if df.empty:
        return df
    sign = df["side"].map({"Buy": 1, "Sell": -1}).fillna(0)
    df["dev_bps"] = ((df["execPrice"] - df["orderPrice"])
                     / df["orderPrice"] * 10_000 * sign)
    return df


def quality_report(execs: pd.DataFrame) -> str:
    df = slippage_vs_mark(execs)
    if df.empty:
        return "評価できる約定がありません"

    lines = ["=== 執行品質（マーク価格に対するスリッページ） ==="]

    def stats(sub: pd.DataFrame, label: str) -> str:
        if sub.empty:
            return f"  {label:<18} データなし"
        s = sub["slip_bps"]
        weighted = ((sub["slip_bps"] * sub["execValue"]).sum()
                    / sub["execValue"].sum())
        return (f"  {label:<18} n={len(s):<5} "
                f"中央値 {s.median():>+7.2f}b  "
                f"平均 {s.mean():>+7.2f}b  "
                f"金額加重 {weighted:>+7.2f}b  "
                f"p90 {s.quantile(0.9):>+7.2f}b")

    lines.append(stats(df, "全体"))
    lines.append(stats(df[df["isMaker"]], "Maker"))
    lines.append(stats(df[~df["isMaker"]], "Taker"))
    lines.append(stats(df[df["side"] == "Buy"], "買い"))
    lines.append(stats(df[df["side"] == "Sell"], "売り"))

    # 金額規模別（大口ほど滑るはず）
    lines.append("\n  --- 約定金額別 ---")
    bins = [0, 1_000, 10_000, 50_000, 1e12]
    labels = ["<1k", "1k-10k", "10k-50k", ">50k"]
    df["bucket"] = pd.cut(df["execValue"], bins=bins, labels=labels)
    for label, sub in df.groupby("bucket", observed=True):
        lines.append(stats(sub, str(label)))

    # 銘柄別（流動性の低い銘柄ほど滑る）
    lines.append("\n  --- 銘柄別（滑りの大きい順・上位8） ---")
    by_sym = (df.groupby("symbol")
              .agg(n=("slip_bps", "count"),
                   median=("slip_bps", "median"),
                   value=("execValue", "sum"))
              .sort_values("median", ascending=False).head(8))
    for sym, r in by_sym.iterrows():
        lines.append(f"  {sym:<14} n={int(r['n']):<5} "
                     f"中央値 {r['median']:>+7.2f}b  "
                     f"金額 {r['value']:>12,.0f}")

    dev = order_price_deviation(execs)
    if not dev.empty:
        lines.append("\n  --- 注文価格からの乖離 ---")
        lines.append(f"  中央値 {dev['dev_bps'].median():+.2f} bps  "
                     f"p90 {dev['dev_bps'].quantile(0.9):+.2f} bps")

    lines.append("\n※ この中央値を記事31のバックテスト（FillModelの")
    lines.append("  slippage_bps）に入れると、検証が現実に近づく。")
    return "\n".join(lines)


def compare_with_backtest_assumption(execs: pd.DataFrame,
                                     assumed_bps: float = 2.0) -> str:
    """バックテストの前提と実測を比較する。"""
    df = slippage_vs_mark(execs)
    taker = df[~df["isMaker"]]
    if taker.empty:
        return "テイカー約定がないため比較できません"

    actual = float(taker["slip_bps"].median())
    weighted = float((taker["slip_bps"] * taker["execValue"]).sum()
                     / taker["execValue"].sum())
    ratio = actual / assumed_bps if assumed_bps else float("inf")

    verdict = ("前提が楽観的です。バックテストの成績は過大評価されています"
               if ratio > 1.5 else
               "前提は妥当です" if ratio > 0.6 else
               "前提が保守的です（悲観側なので安全）")

    return "\n".join([
        "=== バックテスト前提との比較（テイカー） ===",
        f"  バックテストの前提 : {assumed_bps:.2f} bps",
        f"  実測（中央値）     : {actual:.2f} bps",
        f"  実測（金額加重）   : {weighted:.2f} bps",
        f"  比率               : {ratio:.2f}倍",
        f"  判定               : {verdict}",
    ])


if __name__ == "__main__":
    from datetime import datetime, timedelta, timezone

    from client import make_client
    from exec_fetch import fetch_executions

    c = make_client()
    start = datetime.now(timezone.utc) - timedelta(days=60)
    execs = fetch_executions(c.http, start, category="linear", verbose=False)

    if execs.empty:
        raise SystemExit("約定履歴がありません")

    print(quality_report(execs))
    print()
    print(compare_with_backtest_assumption(execs, assumed_bps=2.0))

"""performance.py — 実運用のパフォーマンス指標を算出する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

バックテスト（記事31）と同じ指標を実運用にも適用し、
「検証と実運用が一致しているか」を確認する。
"""
from __future__ import annotations

from dataclasses import dataclass

import numpy as np
import pandas as pd


@dataclass
class PerfMetrics:
    days: float
    trades: int
    net_pnl: float
    total_fee: float
    win_rate: float
    profit_factor: float
    expectancy: float
    sharpe_daily: float
    sortino_daily: float
    max_drawdown: float
    max_dd_duration_days: float
    calmar: float
    best_day: float
    worst_day: float
    fee_share: float          # 手数料が粗利に占める割合

    def report(self) -> str:
        return "\n".join([
            f"  期間             : {self.days:.0f}日",
            f"  トレード数       : {self.trades}",
            f"  純損益           : {self.net_pnl:>+14,.4f}",
            f"  手数料合計       : {self.total_fee:>14,.4f}",
            f"  手数料/粗利      : {self.fee_share:>13.1%}",
            f"  勝率             : {self.win_rate:>13.1%}",
            f"  プロフィットF    : {self.profit_factor:>14.3f}",
            f"  期待値/トレード  : {self.expectancy:>+14,.4f}",
            "",
            f"  Sharpe(日次年率) : {self.sharpe_daily:>14.2f}",
            f"  Sortino(日次年率): {self.sortino_daily:>14.2f}",
            f"  最大DD           : {self.max_drawdown:>+14,.4f}",
            f"  最長DD期間       : {self.max_dd_duration_days:>13.0f}日",
            f"  Calmar           : {self.calmar:>14.2f}",
            f"  最良日 / 最悪日  : {self.best_day:>+10,.2f} / "
            f"{self.worst_day:>+10,.2f}",
        ])


def compute_metrics(pnl: pd.DataFrame, initial_equity: float | None = None
                    ) -> PerfMetrics:
    """closed_pnlのDataFrameから指標を算出する。"""
    if pnl.empty:
        raise ValueError("データがありません")

    s = pnl["net_pnl"]
    fees = float(pnl["total_fee"].sum())
    days = max((pnl.index[-1] - pnl.index[0]).total_seconds() / 86400, 1.0)

    # 日次に集約してリスク指標を計算する
    daily = s.resample("D").sum().fillna(0)
    equity_curve = daily.cumsum()

    # 元本が分かるなら比率で、分からなければ金額ベースで計算する
    if initial_equity and initial_equity > 0:
        rets = daily / initial_equity
    else:
        rets = daily

    std = float(rets.std(ddof=1)) if len(rets) > 1 else 0.0
    sharpe = (float(rets.mean()) / std * np.sqrt(365)) if std > 0 else 0.0

    downside = rets[rets < 0]
    dstd = float(downside.std(ddof=1)) if len(downside) > 1 else 0.0
    sortino = (float(rets.mean()) / dstd * np.sqrt(365)) if dstd > 0 else 0.0

    # ドローダウン（累積損益ベース）
    peak = equity_curve.cummax()
    dd = equity_curve - peak
    max_dd = float(dd.min()) if len(dd) else 0.0

    under = (equity_curve < peak).astype(int)
    longest = run = 0
    for v in under:
        run = run + 1 if v else 0
        longest = max(longest, run)

    wins = s[s > 0]
    losses = s[s < 0]
    gross_win, gross_loss = float(wins.sum()), float(-losses.sum())
    net = float(s.sum())

    annual = net / days * 365
    calmar = (annual / abs(max_dd)) if max_dd < 0 else 0.0

    return PerfMetrics(
        days=days, trades=len(s), net_pnl=net, total_fee=fees,
        win_rate=(len(wins) / len(s)) if len(s) else 0.0,
        profit_factor=(gross_win / gross_loss) if gross_loss else float("inf"),
        expectancy=float(s.mean()),
        sharpe_daily=sharpe, sortino_daily=sortino,
        max_drawdown=max_dd, max_dd_duration_days=float(longest),
        calmar=calmar,
        best_day=float(daily.max()) if len(daily) else 0.0,
        worst_day=float(daily.min()) if len(daily) else 0.0,
        fee_share=(fees / (net + fees)) if (net + fees) else 0.0,
    )


def monthly_table(pnl: pd.DataFrame) -> pd.DataFrame:
    m = pnl["net_pnl"].resample("ME").sum()
    df = m.to_frame("pnl")
    df["year"] = df.index.year
    df["month"] = df.index.month
    return df.pivot_table(index="year", columns="month", values="pnl",
                          fill_value=0)


if __name__ == "__main__":
    from datetime import datetime, timedelta, timezone

    from client import make_client
    from pnl_fetch import fetch_closed_pnl

    c = make_client()
    start = datetime.now(timezone.utc) - timedelta(days=180)
    pnl = fetch_closed_pnl(c.http, start, verbose=False)

    if pnl.empty:
        raise SystemExit("決済履歴がありません")

    bal = c.http.get_wallet_balance(accountType="UNIFIED")["result"]["list"][0]
    equity = float(bal["totalEquity"] or 0)

    print("=== 実運用パフォーマンス ===")
    print(compute_metrics(pnl, initial_equity=equity).report())

    print("\n=== 月次損益 ===")
    print(monthly_table(pnl).to_string(float_format=lambda v: f"{v:,.2f}"))

    print("\n※ この数値をバックテストの結果と比較する。")
    print("  乖離が大きいなら、コストモデルか約定モデルの前提が")
    print("  現実と合っていない（記事31・記事32）。")

"""history_store.py — 約定・損益履歴のSQLiteストア（差分ETL）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

APIは過去2年分しか保持しないため、
定期的に取得してローカルに蓄積しておくのが重要（記事40の確定申告でも使う）。
"""
from __future__ import annotations

import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pandas as pd

DB_PATH = Path("data/history.sqlite")

SCHEMA = """
CREATE TABLE IF NOT EXISTS executions (
    exec_id     TEXT PRIMARY KEY,
    exec_time   INTEGER NOT NULL,
    category    TEXT,
    symbol      TEXT,
    side        TEXT,
    exec_price  REAL,
    exec_qty    REAL,
    exec_value  REAL,
    exec_fee    REAL,
    fee_rate    REAL,
    fee_currency TEXT,
    is_maker    INTEGER,
    exec_type   TEXT,
    exec_pnl    REAL,
    closed_size REAL,
    mark_price  REAL,
    order_price REAL,
    order_type  TEXT,
    order_id    TEXT,
    order_link_id TEXT
);
CREATE INDEX IF NOT EXISTS idx_exec_time ON executions(exec_time);
CREATE INDEX IF NOT EXISTS idx_exec_symbol ON executions(symbol);
CREATE INDEX IF NOT EXISTS idx_exec_link ON executions(order_link_id);

CREATE TABLE IF NOT EXISTS closed_pnl (
    key         TEXT PRIMARY KEY,
    updated_ms  INTEGER NOT NULL,
    category    TEXT,
    symbol      TEXT,
    side        TEXT,
    qty         REAL,
    closed_size REAL,
    avg_entry   REAL,
    avg_exit    REAL,
    closed_pnl  REAL,
    open_fee    REAL,
    close_fee   REAL,
    leverage    REAL,
    order_id    TEXT
);
CREATE INDEX IF NOT EXISTS idx_pnl_time ON closed_pnl(updated_ms);
CREATE INDEX IF NOT EXISTS idx_pnl_symbol ON closed_pnl(symbol);

CREATE TABLE IF NOT EXISTS sync_state (
    name       TEXT PRIMARY KEY,
    last_ms    INTEGER NOT NULL,
    updated_at REAL NOT NULL
);
"""


class HistoryStore:
    def __init__(self, path: Path = DB_PATH):
        path.parent.mkdir(parents=True, exist_ok=True)
        self.conn = sqlite3.connect(str(path), isolation_level=None)
        self.conn.row_factory = sqlite3.Row
        self.conn.execute("PRAGMA journal_mode=WAL")
        self.conn.executescript(SCHEMA)

    # ---- 同期状態 --------------------------------------------------------
    def last_synced_ms(self, name: str, default_days: int = 90) -> int:
        r = self.conn.execute(
            "SELECT last_ms FROM sync_state WHERE name = ?", (name,)
        ).fetchone()
        if r:
            return int(r["last_ms"])
        fallback = datetime.now(timezone.utc) - timedelta(days=default_days)
        return int(fallback.timestamp() * 1000)

    def set_synced_ms(self, name: str, ms: int) -> None:
        import time as _t

        self.conn.execute(
            "INSERT INTO sync_state (name, last_ms, updated_at) VALUES (?,?,?)"
            " ON CONFLICT(name) DO UPDATE SET last_ms=?, updated_at=?",
            (name, ms, _t.time(), ms, _t.time()),
        )

    # ---- 取り込み --------------------------------------------------------
    def upsert_executions(self, df: pd.DataFrame, category: str) -> int:
        if df.empty:
            return 0
        rows = []
        for ts, r in df.iterrows():
            rows.append((
                r.get("execId"), int(r.get("execTime") or 0), category,
                r.get("symbol"), r.get("side"),
                float(r.get("execPrice") or 0), float(r.get("execQty") or 0),
                float(r.get("execValue") or 0), float(r.get("execFee") or 0),
                float(r.get("feeRate") or 0), r.get("feeCurrency"),
                1 if r.get("isMaker") else 0, r.get("execType"),
                float(r.get("execPnl") or 0), float(r.get("closedSize") or 0),
                float(r.get("markPrice") or 0), float(r.get("orderPrice") or 0),
                r.get("orderType"), r.get("orderId"), r.get("orderLinkId"),
            ))
        self.conn.executemany(
            "INSERT OR REPLACE INTO executions VALUES "
            "(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", rows
        )
        return len(rows)

    def upsert_closed_pnl(self, df: pd.DataFrame, category: str) -> int:
        if df.empty:
            return 0
        rows = []
        for ts, r in df.iterrows():
            key = f"{r.get('orderId')}|{r.get('updatedTime')}"
            rows.append((
                key, int(r.get("updatedTime") or 0), category,
                r.get("symbol"), r.get("side"),
                float(r.get("qty") or 0), float(r.get("closedSize") or 0),
                float(r.get("avgEntryPrice") or 0),
                float(r.get("avgExitPrice") or 0),
                float(r.get("closedPnl") or 0),
                float(r.get("openFee") or 0), float(r.get("closeFee") or 0),
                float(r.get("leverage") or 0), r.get("orderId"),
            ))
        self.conn.executemany(
            "INSERT OR REPLACE INTO closed_pnl VALUES "
            "(?,?,?,?,?,?,?,?,?,?,?,?,?,?)", rows
        )
        return len(rows)

    # ---- 差分同期 --------------------------------------------------------
    def sync(self, http, categories=("linear", "spot"),
             default_days: int = 90, verbose: bool = True) -> dict:
        from exec_fetch import fetch_executions
        from pnl_fetch import fetch_closed_pnl

        result = {}
        now = datetime.now(timezone.utc)

        for category in categories:
            name = f"exec_{category}"
            since_ms = self.last_synced_ms(name, default_days)
            # 重複を許して少し戻る（境界の取りこぼしを防ぐ）
            since = datetime.fromtimestamp(since_ms / 1000, timezone.utc)
            since -= timedelta(minutes=5)

            if verbose:
                print(f"[sync] {name}: {since:%Y-%m-%d %H:%M} から")
            df = fetch_executions(http, since, now, category=category,
                                  verbose=False)
            n = self.upsert_executions(df, category)
            result[name] = n
            if not df.empty:
                self.set_synced_ms(name, int(df["execTime"].astype("int64").max()))

            if category in ("linear", "inverse"):
                pname = f"pnl_{category}"
                p_since_ms = self.last_synced_ms(pname, default_days)
                p_since = datetime.fromtimestamp(p_since_ms / 1000,
                                                 timezone.utc)
                p_since -= timedelta(minutes=5)
                pdf = fetch_closed_pnl(http, p_since, now, category=category,
                                       verbose=False)
                result[pname] = self.upsert_closed_pnl(pdf, category)
                if not pdf.empty:
                    self.set_synced_ms(
                        pname, int(pdf["updatedTime"].astype("int64").max()))

        if verbose:
            print(f"[sync] 完了: {result}")
        return result

    # ---- 参照 ------------------------------------------------------------
    def executions(self, since: str | None = None) -> pd.DataFrame:
        q = "SELECT * FROM executions"
        params: tuple = ()
        if since:
            ms = int(pd.Timestamp(since, tz="UTC").timestamp() * 1000)
            q += " WHERE exec_time >= ?"
            params = (ms,)
        q += " ORDER BY exec_time"
        df = pd.read_sql_query(q, self.conn, params=params)
        if df.empty:
            return df
        df["ts"] = pd.to_datetime(df["exec_time"], unit="ms", utc=True)
        return df.set_index("ts")

    def summary(self) -> str:
        e = self.conn.execute(
            "SELECT COUNT(*) n, MIN(exec_time) a, MAX(exec_time) b,"
            " SUM(exec_fee) fee FROM executions").fetchone()
        p = self.conn.execute(
            "SELECT COUNT(*) n, SUM(closed_pnl) pnl FROM closed_pnl"
        ).fetchone()

        def fmt(ms) -> str:
            if not ms:
                return "-"
            return f"{datetime.fromtimestamp(ms / 1000, timezone.utc):%Y-%m-%d}"

        return "\n".join([
            f"約定      : {e['n']}件  {fmt(e['a'])} 〜 {fmt(e['b'])}",
            f"手数料合計 : {(e['fee'] or 0):,.4f}",
            f"決済      : {p['n']}件  純損益 {(p['pnl'] or 0):,.4f}",
        ])


if __name__ == "__main__":
    from client import make_client

    c = make_client()
    store = HistoryStore()
    store.sync(c.http, categories=("linear", "spot"), default_days=90)
    print()
    print(store.summary())
    print("\n※ cronで日次実行すれば、常に最新の履歴が手元に揃う（記事38）。")
    print("  APIは過去2年しか保持しないため、蓄積しておく価値がある。")

"""export_report.py — 集計結果をCSV/Excelに出力する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

用途: 手元での確認、税務用の元データ（記事40）、
      スプレッドシートでの追加分析。
"""
from __future__ import annotations

from datetime import datetime, timedelta, timezone
from pathlib import Path

import pandas as pd

OUT_DIR = Path("reports")


def export_all(execs: pd.DataFrame, pnl: pd.DataFrame,
               out_dir: Path = OUT_DIR, known_strategies=None) -> list[Path]:
    """複数の集計をまとめて出力する。"""
    from exec_quality import slippage_vs_mark
    from fee_analysis import fee_by_symbol, fee_timeline
    from strategy_pnl import strategy_report

    out_dir.mkdir(parents=True, exist_ok=True)
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d")
    written: list[Path] = []

    sheets: dict[str, pd.DataFrame] = {}

    # 生データ（列を絞って読みやすくする）
    if not execs.empty:
        cols = ["symbol", "side", "execType", "execPrice", "execQty",
                "execValue", "execFee", "feeRate", "isMaker", "execPnl",
                "orderLinkId", "orderId"]
        sheets["約定明細"] = execs[[c for c in cols if c in execs.columns]]
        sheets["銘柄別手数料"] = fee_by_symbol(execs, top=100)
        sheets["週次手数料"] = fee_timeline(execs, "W")
        slip = slippage_vs_mark(execs)
        if not slip.empty:
            sheets["執行品質"] = slip[["symbol", "side", "execValue",
                                       "isMaker", "slip_bps"]]

    if not pnl.empty:
        cols = ["symbol", "side", "qty", "avgEntryPrice", "avgExitPrice",
                "closedPnl", "openFee", "closeFee", "leverage"]
        sheets["決済明細"] = pnl[[c for c in cols if c in pnl.columns]]

    if not execs.empty:
        sheets["戦略別"] = strategy_report(execs, pnl, known_strategies)

    # Excelにまとめる（openpyxlが無い環境ではCSVにフォールバック）
    xlsx = out_dir / f"bybit_report_{stamp}.xlsx"
    try:
        with pd.ExcelWriter(xlsx, engine="openpyxl") as writer:
            for name, df in sheets.items():
                # Excelはタイムゾーン付きindexを扱えないので落とす
                out = df.copy()
                if isinstance(out.index, pd.DatetimeIndex):
                    out.index = out.index.tz_localize(None)
                out.to_excel(writer, sheet_name=name[:31])
        written.append(xlsx)
        print(f"[出力] {xlsx}")
    except ImportError:
        print("[info] openpyxlが無いためCSVで出力します")
        for name, df in sheets.items():
            p = out_dir / f"{stamp}_{name}.csv"
            df.to_csv(p, encoding="utf-8-sig")   # Excelで開ける
            written.append(p)
            print(f"[出力] {p}")

    return written


if __name__ == "__main__":
    from client import make_client
    from exec_fetch import fetch_executions
    from pnl_fetch import fetch_closed_pnl

    c = make_client()
    start = datetime.now(timezone.utc) - timedelta(days=90)

    execs = fetch_executions(c.http, start, category="linear", verbose=False)
    pnl = fetch_closed_pnl(c.http, start, verbose=False)

    if execs.empty and pnl.empty:
        raise SystemExit("出力するデータがありません")

    export_all(execs, pnl, known_strategies={"trendfol", "gridbot", "carry"})
    print("\n※ 年間の損益集計と確定申告用のデータ作成は記事40で扱う。")