Bybitの公開ティックデータで長期の約定履歴を取得する
情報駆動バーの生成とPython実装

ベース: https://public.bybit.com/trading/

  https://public.bybit.com/trading/{SYMBOL}/
    -> その銘柄の日次ファイル一覧

  https://public.bybit.com/trading/{SYMBOL}/{SYMBOL}{YYYY-MM-DD}.csv.gz
    -> 1日分の全約定（gzip圧縮CSV）

例:
  https://public.bybit.com/trading/BTCUSDT/BTCUSDT2025-01-09.csv.gz

利用できる銘柄:
  - USDT無期限（BTCUSDT, ETHUSDT, ...）
  - USDC無期限（BTCPERP, ...）
  - コイン建て（BTCUSD, ...）
  - 日付付き先物（BTC-27JUN25 など）
  - 現物ペア

BTCUSDTは2020年3月25日から提供されています。
銘柄によって開始日は異なります（上場日以降）。

"""public_data.py — public.bybit.com から日次ティックCSVを取得する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

特徴:
  - 並列ダウンロード（ただし相手はCDNなので節度を持つ）
  - レジューム（既にあるファイルはスキップ）
  - 存在しない日付（休止期間など）を記録して再試行しない
"""
from __future__ import annotations

import json
import re
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from pathlib import Path

import requests

BASE = "https://public.bybit.com/trading"
RAW_DIR = Path("data/public/raw")
STATE_PATH = Path("data/public/download_state.json")

# 相手はCDNだが、節度を持った並列数にする
DEFAULT_WORKERS = 4
TIMEOUT = 120


@dataclass
class DownloadStats:
    symbol: str
    downloaded: int = 0
    skipped: int = 0
    missing: int = 0
    failed: list[str] = field(default_factory=list)
    bytes_total: int = 0

    def report(self) -> str:
        mb = self.bytes_total / 1024 / 1024
        return (f"{self.symbol}: 新規 {self.downloaded} / "
                f"既存 {self.skipped} / 欠番 {self.missing} / "
                f"失敗 {len(self.failed)}  ({mb:,.1f} MB)")


def list_available_dates(symbol: str, timeout: int = 30) -> list[date]:
    """銘柄ディレクトリのHTMLから、利用できる日付を抽出する。"""
    url = f"{BASE}/{symbol}/"
    r = requests.get(url, timeout=timeout)
    r.raise_for_status()

    # ファイル名: {SYMBOL}{YYYY-MM-DD}.csv.gz
    pattern = re.compile(rf"{re.escape(symbol)}(\d{{4}}-\d{{2}}-\d{{2}})\.csv\.gz")
    dates = sorted({
        datetime.strptime(m, "%Y-%m-%d").date()
        for m in pattern.findall(r.text)
    })
    return dates


def daterange(start: date, end: date):
    cur = start
    while cur <= end:
        yield cur
        cur += timedelta(days=1)


class PublicDataDownloader:
    def __init__(self, raw_dir: Path = RAW_DIR,
                 state_path: Path = STATE_PATH,
                 workers: int = DEFAULT_WORKERS):
        self.raw_dir = raw_dir
        self.state_path = state_path
        self.workers = workers
        self.raw_dir.mkdir(parents=True, exist_ok=True)
        self._state = self._load_state()

    # ---- 欠番の記録（再試行を避ける） -----------------------------------
    def _load_state(self) -> dict:
        if self.state_path.exists():
            try:
                return json.loads(self.state_path.read_text(encoding="utf-8"))
            except Exception:
                pass
        return {"missing": {}}

    def _save_state(self) -> None:
        self.state_path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.state_path.with_suffix(".tmp")
        tmp.write_text(json.dumps(self._state), encoding="utf-8")
        tmp.replace(self.state_path)

    def _is_known_missing(self, symbol: str, d: date) -> bool:
        return str(d) in self._state["missing"].get(symbol, [])

    def _mark_missing(self, symbol: str, d: date) -> None:
        self._state["missing"].setdefault(symbol, [])
        if str(d) not in self._state["missing"][symbol]:
            self._state["missing"][symbol].append(str(d))

    # ---- ダウンロード ----------------------------------------------------
    def path_for(self, symbol: str, d: date) -> Path:
        return self.raw_dir / symbol / f"{symbol}{d:%Y-%m-%d}.csv.gz"

    def _download_one(self, symbol: str, d: date,
                      session: requests.Session) -> tuple[str, int]:
        """戻り値: (状態, バイト数)。状態は ok / skip / missing / fail。"""
        path = self.path_for(symbol, d)
        if path.exists() and path.stat().st_size > 0:
            return "skip", path.stat().st_size
        if self._is_known_missing(symbol, d):
            return "missing", 0

        url = f"{BASE}/{symbol}/{symbol}{d:%Y-%m-%d}.csv.gz"
        try:
            with session.get(url, stream=True, timeout=TIMEOUT) as r:
                if r.status_code == 404:
                    self._mark_missing(symbol, d)
                    return "missing", 0
                r.raise_for_status()

                path.parent.mkdir(parents=True, exist_ok=True)
                tmp = path.with_suffix(".part")
                size = 0
                with tmp.open("wb") as f:
                    for chunk in r.iter_content(chunk_size=1 << 20):
                        f.write(chunk)
                        size += len(chunk)
                tmp.replace(path)          # 部分ファイルを残さない
                return "ok", size
        except Exception:
            return "fail", 0

    def download(self, symbol: str, start: date, end: date,
                 verbose: bool = True) -> DownloadStats:
        stats = DownloadStats(symbol=symbol)
        days = list(daterange(start, end))

        session = requests.Session()
        with ThreadPoolExecutor(max_workers=self.workers) as pool:
            futures = {pool.submit(self._download_one, symbol, d, session): d
                       for d in days}
            done = 0
            for fut in as_completed(futures):
                d = futures[fut]
                status, size = fut.result()
                done += 1

                if status == "ok":
                    stats.downloaded += 1
                    stats.bytes_total += size
                elif status == "skip":
                    stats.skipped += 1
                    stats.bytes_total += size
                elif status == "missing":
                    stats.missing += 1
                else:
                    stats.failed.append(str(d))

                if verbose and done % 50 == 0:
                    print(f"  {done}/{len(days)} … "
                          f"{stats.bytes_total / 1024 / 1024:,.0f} MB")

        self._save_state()
        if verbose:
            print(stats.report())
        return stats

    def estimate_size(self, symbol: str, sample_days: int = 3) -> float:
        """1日あたりの平均サイズ(MB)を推定する。取得前の見積り用。"""
        files = sorted((self.raw_dir / symbol).glob("*.csv.gz"))[-sample_days:]
        if not files:
            return 0.0
        return sum(f.stat().st_size for f in files) / len(files) / 1024 / 1024


if __name__ == "__main__":
    SYMBOL = "BTCUSDT"

    print(f"=== {SYMBOL} の利用可能な期間を確認 ===")
    dates = list_available_dates(SYMBOL)
    print(f"  {len(dates)}日分  {dates[0]} 〜 {dates[-1]}")

    dl = PublicDataDownloader()

    # まず1週間だけ取ってサイズを見る
    print(f"\n=== 直近7日を取得してサイズを確認 ===")
    end = dates[-1]
    dl.download(SYMBOL, end - timedelta(days=6), end)

    per_day = dl.estimate_size(SYMBOL)
    print(f"\n1日あたり約 {per_day:.1f} MB")
    print(f"  1年分なら約 {per_day * 365 / 1024:.1f} GB")
    print(f"  全期間({len(dates)}日)なら約 "
          f"{per_day * len(dates) / 1024:.1f} GB")
    print("\n※ ディスク容量を確認してから全期間の取得を始めてください。")

"""tick_parser.py — 日次CSVをパースしてParquetに変換する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

要点:
  - 列名を決め打ちせず、ヘッダから実際の列を読む
  - dtypeを明示してメモリを節約する
  - 変換後は生CSVを削除できる（Parquetは1/5〜1/10）
"""
from __future__ import annotations

import gzip
from dataclasses import dataclass
from datetime import date
from pathlib import Path

import pandas as pd

PARQUET_DIR = Path("data/public/parquet")

# 想定される列名の候補（時期・銘柄で揺れるため複数を許容する）
COLUMN_ALIASES = {
    "ts": ["timestamp", "trdTime", "time"],
    "price": ["price"],
    "size": ["size", "qty", "volume"],
    "side": ["side"],
    "tick_direction": ["tickDirection", "tick_direction"],
    "trade_id": ["trdMatchID", "tradeId", "id"],
    "gross_value": ["grossValue"],
    "home_notional": ["homeNotional"],
    "foreign_notional": ["foreignNotional"],
}


@dataclass
class ParseResult:
    path: Path
    rows: int
    columns: list[str]
    unmapped: list[str]

    def report(self) -> str:
        base = f"{self.path.name}: {self.rows:,}行 / 列 {len(self.columns)}"
        if self.unmapped:
            base += f"  未マップ列: {self.unmapped}"
        return base


def peek_header(csv_gz: Path) -> list[str]:
    """ファイルのヘッダ行だけを読む（列構成の確認用）。"""
    with gzip.open(csv_gz, "rt", encoding="utf-8") as f:
        header = f.readline().strip()
    return [c.strip() for c in header.split(",")]


def build_rename_map(columns: list[str]) -> tuple[dict[str, str], list[str]]:
    """実際の列名から、標準名へのマッピングを作る。"""
    lower = {c.lower(): c for c in columns}
    rename: dict[str, str] = {}
    for std, aliases in COLUMN_ALIASES.items():
        for a in aliases:
            if a.lower() in lower:
                rename[lower[a.lower()]] = std
                break
    unmapped = [c for c in columns if c not in rename]
    return rename, unmapped


def parse_csv(csv_gz: Path, chunksize: int | None = None) -> pd.DataFrame:
    """日次CSVをDataFrameに読み込む。"""
    columns = peek_header(csv_gz)
    rename, _ = build_rename_map(columns)

    # メモリを節約するためdtypeを指定する
    dtypes: dict[str, str] = {}
    for col in columns:
        std = rename.get(col)
        if std in ("price", "size", "gross_value", "home_notional",
                   "foreign_notional"):
            dtypes[col] = "float64"
        elif std in ("side", "tick_direction", "trade_id"):
            dtypes[col] = "string"

    df = pd.read_csv(csv_gz, compression="gzip", dtype=dtypes,
                     low_memory=False)
    df = df.rename(columns=rename)

    if "ts" not in df.columns:
        raise ValueError(f"{csv_gz.name}: timestamp列が見つかりません "
                         f"(列: {columns})")

    # timestampは秒（小数付き）で入っていることが多い。
    # 桁数から秒/ミリ秒を自動判定する
    ts = pd.to_numeric(df["ts"], errors="coerce")
    unit = "ms" if ts.max() > 1e12 else "s"
    df["ts"] = pd.to_datetime(ts, unit=unit, utc=True)

    # 売買方向は文字列。カテゴリ型にしてメモリを削る
    for col in ("side", "tick_direction"):
        if col in df.columns:
            df[col] = df[col].astype("category")

    return df.set_index("ts").sort_index()


def to_parquet(csv_gz: Path, out_dir: Path = PARQUET_DIR,
               delete_source: bool = False) -> ParseResult:
    """日次CSVをParquetに変換する。"""
    symbol = csv_gz.parent.name
    out_sym = out_dir / symbol
    out_sym.mkdir(parents=True, exist_ok=True)

    day = csv_gz.stem.replace(symbol, "").replace(".csv", "")
    out = out_sym / f"{symbol}_{day}.parquet"

    if out.exists() and out.stat().st_size > 0:
        return ParseResult(out, 0, [], [])

    df = parse_csv(csv_gz)
    columns = peek_header(csv_gz)
    _, unmapped = build_rename_map(columns)

    tmp = out.with_suffix(".tmp")
    df.to_parquet(tmp, compression="zstd")
    tmp.replace(out)

    if delete_source:
        csv_gz.unlink()

    return ParseResult(out, len(df), list(df.columns), unmapped)


def convert_all(symbol: str, raw_dir: Path = Path("data/public/raw"),
                out_dir: Path = PARQUET_DIR,
                delete_source: bool = False,
                verbose: bool = True) -> dict:
    files = sorted((raw_dir / symbol).glob("*.csv.gz"))
    if not files:
        return {"error": f"{symbol}: CSVが見つかりません"}

    total_rows = 0
    converted = 0
    csv_bytes = 0
    parquet_bytes = 0

    for i, f in enumerate(files, 1):
        csv_bytes += f.stat().st_size
        res = to_parquet(f, out_dir, delete_source)
        if res.rows:
            converted += 1
            total_rows += res.rows
        parquet_bytes += res.path.stat().st_size if res.path.exists() else 0

        if verbose and i % 20 == 0:
            print(f"  {i}/{len(files)} … {total_rows:,}行")

    ratio = (parquet_bytes / csv_bytes) if csv_bytes else 0
    return {
        "symbol": symbol,
        "ファイル数": len(files),
        "変換": converted,
        "総ティック数": total_rows,
        "CSV(MB)": round(csv_bytes / 1024 / 1024, 1),
        "Parquet(MB)": round(parquet_bytes / 1024 / 1024, 1),
        "圧縮率": f"{ratio:.1%}",
    }


def load_range(symbol: str, start: date, end: date,
               parquet_dir: Path = PARQUET_DIR) -> pd.DataFrame:
    """期間を指定してParquetを読み込む。"""
    files = []
    for p in sorted((parquet_dir / symbol).glob("*.parquet")):
        day_str = p.stem.split("_")[-1]
        try:
            d = date.fromisoformat(day_str)
        except ValueError:
            continue
        if start <= d <= end:
            files.append(p)

    if not files:
        return pd.DataFrame()
    return pd.concat([pd.read_parquet(p) for p in files]).sort_index()


if __name__ == "__main__":
    SYMBOL = "BTCUSDT"
    raw = Path("data/public/raw") / SYMBOL
    files = sorted(raw.glob("*.csv.gz"))

    if not files:
        raise SystemExit("先に public_data.py でダウンロードしてください")

    print("=== ファイルの列構成を確認 ===")
    cols = peek_header(files[-1])
    print(f"  {files[-1].name}")
    print(f"  列: {cols}")
    rename, unmapped = build_rename_map(cols)
    print(f"  マップ: {rename}")
    if unmapped:
        print(f"  未マップ（無視される列）: {unmapped}")

    print("\n=== Parquetへ変換 ===")
    for k, v in convert_all(SYMBOL).items():
        print(f"  {k:<14}: {v}")

    print("\n=== 読み込みテスト ===")
    df = load_range(SYMBOL, files[0].stem[-10:] and
                    date.fromisoformat(files[-1].stem.replace(SYMBOL, "")
                                       .replace(".csv", "")),
                    date.fromisoformat(files[-1].stem.replace(SYMBOL, "")
                                       .replace(".csv", "")))
    if not df.empty:
        print(f"  {len(df):,}ティック  {df.index[0]} 〜 {df.index[-1]}")
        print(df.head(3).to_string())

"""recent_ticks.py — 直近のティックをAPI/WSで補完する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

役割分担:
  過去（前日まで） -> public.bybit.com（サンプル1・2）
  直近            -> get_public_trade_history（1000件程度）
  リアルタイム     -> WebSocket publicTrade（記事26）
"""
from __future__ import annotations

import time
from pathlib import Path

import pandas as pd

RECENT_DIR = Path("data/public/recent")


def fetch_recent_trades(http, symbol: str, category: str = "linear",
                        limit: int = 1000) -> pd.DataFrame:
    """APIで直近の約定を取得する（取得できるのは直近のみ）。"""
    res = http.get_public_trade_history(category=category, symbol=symbol,
                                        limit=min(limit, 1000))
    if res["retCode"] != 0:
        raise RuntimeError(f"get_public_trade_history失敗: {res['retMsg']}")

    rows = res["result"]["list"]
    if not rows:
        return pd.DataFrame()

    df = pd.DataFrame([{
        "ts": pd.Timestamp(int(r["time"]), unit="ms", tz="UTC"),
        "price": float(r["price"]),
        "size": float(r["size"]),
        "side": r["side"],              # テイカーの方向
        "trade_id": r.get("execId", ""),
    } for r in rows])
    return df.set_index("ts").sort_index()


class TickRecorder:
    """WebSocketのpublicTradeを日次ファイルに記録する（記事26と連携）。

    公開データが翌日まで出ないため、当日分を自分で貯めておく。
    """

    def __init__(self, symbol: str, out_dir: Path = RECENT_DIR,
                 flush_every: int = 2000):
        self.symbol = symbol
        self.out_dir = out_dir
        self.out_dir.mkdir(parents=True, exist_ok=True)
        self.flush_every = flush_every
        self._buf: list[dict] = []
        self._seen: set[str] = set()
        self.total = 0

    def on_message(self, msg: dict) -> None:
        """WSコールバック。重い処理はしない（記事26の原則）。"""
        for t in msg.get("data", []):
            tid = t.get("i", "")
            if tid and tid in self._seen:
                continue
            if tid:
                self._seen.add(tid)
                if len(self._seen) > 100_000:
                    self._seen.clear()

            self._buf.append({
                "ts_ms": int(t["T"]), "price": float(t["p"]),
                "size": float(t["v"]), "side": t["S"],
                "tick_direction": t.get("L", ""), "trade_id": tid,
                "block_trade": bool(t.get("BT", False)),
            })

        if len(self._buf) >= self.flush_every:
            self.flush()

    def flush(self) -> None:
        if not self._buf:
            return
        df = pd.DataFrame(self._buf)
        df["ts"] = pd.to_datetime(df["ts_ms"], unit="ms", utc=True)
        df = df.drop(columns=["ts_ms"]).set_index("ts").sort_index()

        day = df.index[0].strftime("%Y-%m-%d")
        path = self.out_dir / f"{self.symbol}_{day}.parquet"

        if path.exists():
            df = pd.concat([pd.read_parquet(path), df])
            df = df[~df.index.duplicated(keep="last")].sort_index()

        tmp = path.with_suffix(".tmp")
        df.to_parquet(tmp, compression="zstd")
        tmp.replace(path)

        self.total += len(self._buf)
        self._buf.clear()


def stitch(historical: pd.DataFrame, recent: pd.DataFrame) -> pd.DataFrame:
    """公開データと直近データを継ぎ目なくつなぐ。"""
    if historical.empty:
        return recent
    if recent.empty:
        return historical

    # 重複期間は公開データを優先する（確定値のため）
    cutoff = historical.index[-1]
    tail = recent[recent.index > cutoff]
    return pd.concat([historical, tail]).sort_index()


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

    from datetime import date, timedelta

    c = make_client()
    SYMBOL = "BTCUSDT"

    recent = fetch_recent_trades(c.http, SYMBOL)
    print(f"APIで取得した直近ティック: {len(recent)}件")
    if not recent.empty:
        print(f"  {recent.index[0]} 〜 {recent.index[-1]}")
        span = (recent.index[-1] - recent.index[0]).total_seconds()
        print(f"  カバー範囲: わずか {span:.0f}秒分")
        print("  -> 長期の分析には public.bybit.com が必須")

    yesterday = date.today() - timedelta(days=1)
    hist = load_range(SYMBOL, yesterday - timedelta(days=2), yesterday)
    if not hist.empty:
        merged = stitch(hist, recent)
        print(f"\n結合後: {len(merged):,}ティック")
        print(f"  {merged.index[0]} 〜 {merged.index[-1]}")

    # WebSocketでの記録（当日分を自分で貯める）
    print("\n=== WebSocketで記録する場合 ===")
    print("  from pybit.unified_trading import WebSocket")
    print("  rec = TickRecorder('BTCUSDT')")
    print("  ws.trade_stream(symbol='BTCUSDT', callback=rec.on_message)")

"""information_bars.py — 情報駆動バーの生成
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

参考: Lopez de Prado (2018)
      "Advances in Financial Machine Learning" Ch.2
"""
from __future__ import annotations

from dataclasses import dataclass

import numpy as np
import pandas as pd


@dataclass
class BarSpec:
    kind: str            # tick / volume / dollar
    threshold: float

    def describe(self) -> str:
        unit = {"tick": "約定", "volume": "枚", "dollar": "USDT"}[self.kind]
        return f"{self.kind}バー（{self.threshold:,.0f} {unit}ごと）"


def _accumulator(df: pd.DataFrame, kind: str) -> np.ndarray:
    if kind == "tick":
        return np.ones(len(df), dtype="float64")
    if kind == "volume":
        return df["size"].to_numpy(dtype="float64")
    if kind == "dollar":
        return (df["price"] * df["size"]).to_numpy(dtype="float64")
    raise ValueError(f"未知のバー種別: {kind}")


def make_bars(ticks: pd.DataFrame, spec: BarSpec) -> pd.DataFrame:
    """ティックから情報駆動バーを生成する。

    ticks は index=UTC時刻、列に price / size / side が必要。
    """
    if ticks.empty:
        return pd.DataFrame()

    df = ticks.sort_index()
    acc = _accumulator(df, spec.kind)
    price = df["price"].to_numpy(dtype="float64")
    size = df["size"].to_numpy(dtype="float64")
    is_buy = (df["side"].astype(str) == "Buy").to_numpy()
    ts = df.index.to_numpy()

    bars = []
    cum = 0.0
    start = 0

    for i in range(len(df)):
        cum += acc[i]
        if cum < spec.threshold:
            continue

        seg = slice(start, i + 1)
        p = price[seg]
        v = size[seg]
        b = is_buy[seg]
        notional = p * v

        bars.append({
            "ts": ts[i],
            "ts_open": ts[start],
            "open": p[0], "high": p.max(), "low": p.min(), "close": p[-1],
            "volume": v.sum(),
            "dollar": notional.sum(),
            "ticks": i - start + 1,
            "vwap": notional.sum() / v.sum() if v.sum() else p[-1],
            # テイカー方向の分解（需給の偏り）
            "buy_volume": v[b].sum(),
            "sell_volume": v[~b].sum(),
            "duration_sec": (ts[i] - ts[start]) / np.timedelta64(1, "s"),
        })
        cum = 0.0
        start = i + 1

    if not bars:
        return pd.DataFrame()

    out = pd.DataFrame(bars).set_index("ts")
    out["delta"] = out["buy_volume"] - out["sell_volume"]
    out["delta_ratio"] = out["delta"] / out["volume"].replace(0, np.nan)
    return out


def suggest_threshold(ticks: pd.DataFrame, kind: str,
                      target_bars_per_day: int = 288) -> float:
    """1日あたり目標本数になる閾値を推定する。

    288本 = 5分足相当。まずこの水準から始めると比較しやすい。
    """
    if ticks.empty:
        return 0.0
    acc = _accumulator(ticks, kind)
    days = max((ticks.index[-1] - ticks.index[0]).total_seconds() / 86400,
               1 / 24)
    return float(acc.sum() / days / target_bars_per_day)


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

    from tick_parser import load_range

    SYMBOL = "BTCUSDT"
    end = date.today() - timedelta(days=1)
    ticks = load_range(SYMBOL, end - timedelta(days=6), end)

    if ticks.empty:
        raise SystemExit("先に public_data.py と tick_parser.py を実行してください")

    print(f"{len(ticks):,}ティック  {ticks.index[0]} 〜 {ticks.index[-1]}\n")

    for kind in ("tick", "volume", "dollar"):
        th = suggest_threshold(ticks, kind, target_bars_per_day=288)
        spec = BarSpec(kind=kind, threshold=th)
        bars = make_bars(ticks, spec)

        if bars.empty:
            print(f"{spec.describe()}: 生成できませんでした")
            continue

        days = (ticks.index[-1] - ticks.index[0]).total_seconds() / 86400
        print(f"{spec.describe()}")
        print(f"  本数        : {len(bars):,}（1日あたり {len(bars)/days:.0f}）")
        print(f"  平均所要時間 : {bars['duration_sec'].mean():.0f}秒 "
              f"（最短 {bars['duration_sec'].min():.0f}秒 / "
              f"最長 {bars['duration_sec'].max():.0f}秒）")
        print(f"  平均ティック数: {bars['ticks'].mean():.0f}")
        print(f"  デルタの平均  : {bars['delta_ratio'].mean():+.3f}")
        print()

    print("※ ドルバーは所要時間のばらつきが大きい＝")
    print("  『相場が動くときは細かく、静かなときは粗く』区切れている。")

"""bar_stats.py — タイムバーと情報駆動バーの統計的性質を比較する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

比較する性質:
  1) 正規性     — Jarque-Bera統計量（小さいほど正規に近い）
  2) 系列相関   — Ljung-Box（小さいほど独立に近い）
  3) 分散の安定 — 期間を分割したときの分散のばらつき
"""
from __future__ import annotations

from dataclasses import dataclass

import numpy as np
import pandas as pd
from scipy import stats


@dataclass
class BarQuality:
    label: str
    n_bars: int
    jarque_bera: float
    jb_pvalue: float
    skew: float
    kurtosis: float
    acf1: float
    ljung_box_p: float
    var_stability: float      # 分散のばらつき（小さいほど安定）

    def line(self) -> str:
        return (f"{self.label:<22}{self.n_bars:>7}"
                f"{self.jarque_bera:>12.1f}{self.skew:>+9.3f}"
                f"{self.kurtosis:>9.2f}{self.acf1:>+9.3f}"
                f"{self.ljung_box_p:>9.3f}{self.var_stability:>10.3f}")


def ljung_box(x: np.ndarray, lags: int = 10) -> float:
    """Ljung-Box検定のp値（自作・statsmodels不要）。

    p値が大きいほど「系列相関がない」という帰無仮説が棄却されない。
    """
    n = len(x)
    if n < lags + 5:
        return float("nan")
    x = x - x.mean()
    denom = (x ** 2).sum()
    if denom == 0:
        return float("nan")

    q = 0.0
    for k in range(1, lags + 1):
        r = (x[k:] * x[:-k]).sum() / denom
        q += r ** 2 / (n - k)
    q *= n * (n + 2)
    return float(1 - stats.chi2.cdf(q, df=lags))


def variance_stability(x: np.ndarray, blocks: int = 10) -> float:
    """期間をブロックに分けたときの分散の変動係数。

    小さいほど「どの期間でも分散が同じ」＝安定している。
    """
    if len(x) < blocks * 10:
        return float("nan")
    chunks = np.array_split(x, blocks)
    variances = np.array([c.var(ddof=1) for c in chunks if len(c) > 1])
    if variances.size < 2 or variances.mean() == 0:
        return float("nan")
    return float(variances.std(ddof=1) / variances.mean())


def assess(returns: np.ndarray, label: str) -> BarQuality:
    r = returns[np.isfinite(returns)]
    if len(r) < 30:
        raise ValueError(f"{label}: サンプル不足 ({len(r)})")

    jb, jb_p = stats.jarque_bera(r)
    acf1 = float(np.corrcoef(r[:-1], r[1:])[0, 1]) if len(r) > 2 else 0.0

    return BarQuality(
        label=label, n_bars=len(r),
        jarque_bera=float(jb), jb_pvalue=float(jb_p),
        skew=float(stats.skew(r)),
        kurtosis=float(stats.kurtosis(r, fisher=False)),
        acf1=acf1,
        ljung_box_p=ljung_box(r),
        var_stability=variance_stability(r),
    )


def bar_returns(bars: pd.DataFrame) -> np.ndarray:
    return np.log(bars["close"] / bars["close"].shift(1)).dropna().to_numpy()


def time_bars_from_ticks(ticks: pd.DataFrame, freq: str) -> pd.DataFrame:
    """ティックからタイムバーを作る（比較の基準）。"""
    agg = ticks.resample(freq).agg(
        open=("price", "first"), high=("price", "max"),
        low=("price", "min"), close=("price", "last"),
        volume=("size", "sum"),
    ).dropna(subset=["open"])
    return agg


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

    from information_bars import BarSpec, make_bars, suggest_threshold
    from tick_parser import load_range

    SYMBOL = "BTCUSDT"
    end = date.today() - timedelta(days=1)
    ticks = load_range(SYMBOL, end - timedelta(days=13), end)

    if ticks.empty:
        raise SystemExit("先にデータを取得してください")

    print(f"{len(ticks):,}ティック / "
          f"{(ticks.index[-1] - ticks.index[0]).days}日分\n")

    results: list[BarQuality] = []

    # タイムバー（比較の基準）
    for freq, label in (("5min", "タイムバー(5分)"),
                        ("15min", "タイムバー(15分)"),
                        ("1h", "タイムバー(1時間)")):
        tb = time_bars_from_ticks(ticks, freq)
        if len(tb) > 40:
            results.append(assess(bar_returns(tb), label))

    # 情報駆動バー（本数を揃えて公平に比較する）
    for kind in ("tick", "volume", "dollar"):
        th = suggest_threshold(ticks, kind, target_bars_per_day=288)
        bars = make_bars(ticks, BarSpec(kind, th))
        if len(bars) > 40:
            results.append(assess(bar_returns(bars), f"{kind}バー"))

    print(f"{'バー種別':<22}{'本数':>7}{'JB統計量':>12}{'歪度':>9}"
          f"{'尖度':>9}{'ACF(1)':>9}{'LB p値':>9}{'分散安定':>10}")
    print("-" * 88)
    for r in results:
        print(r.line())

    print("\n--- 読み方 ---")
    print("  JB統計量が小さい  = 正規分布に近い（統計手法が使いやすい）")
    print("  尖度が3に近い     = 極端な値が少ない")
    print("  ACF(1)が0に近い   = 系列相関が小さい（独立に近い）")
    print("  LB p値が大きい    = 系列相関の帰無仮説が棄却されない")
    print("  分散安定が小さい  = どの期間でもばらつきが同程度")
    print()
    print("  一般にドルバーはタイムバーより正規性が高く分散も安定する。")
    print("  ただし『必ずそうなる』わけではないので、自分のデータで確認すること。")

"""trade_direction.py — 売買方向の推定と、記録された方向との照合
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

参考: Lee, C. & Ready, M. (1991)
      "Inferring Trade Direction from Intraday Data"

Bybitのデータには side が含まれるため推定は不要だが、
  - 推定の精度を検証できる（貴重な検証機会）
  - 方向が無い他のデータソースに応用できる
という価値がある。
"""
from __future__ import annotations

import numpy as np
import pandas as pd


def tick_rule(prices: pd.Series) -> pd.Series:
    """ティックルール。前回より高ければ買い、安ければ売り。

    同値の場合は直前の方向を引き継ぐ（zero tick）。
    """
    diff = prices.diff()
    sign = np.sign(diff)
    # 0（同値）は直前の符号を引き継ぐ
    sign = sign.replace(0, np.nan).ffill().fillna(1)
    return sign.astype("int8")


def quote_rule(prices: pd.Series, bid: pd.Series,
               ask: pd.Series) -> pd.Series:
    """クオートルール。ミッドより上なら買い、下なら売り。

    板のスナップショットが必要（記事09・28で取得できる）。
    """
    mid = (bid + ask) / 2
    sign = pd.Series(0, index=prices.index, dtype="int8")
    sign[prices > mid] = 1
    sign[prices < mid] = -1
    return sign


def lee_ready(prices: pd.Series, bid: pd.Series | None = None,
              ask: pd.Series | None = None) -> pd.Series:
    """Lee-Readyアルゴリズム。

    クオートルールを優先し、ミッド価格ちょうどの場合のみ
    ティックルールで補う。板が無ければティックルールのみ。
    """
    tick = tick_rule(prices)
    if bid is None or ask is None:
        return tick

    quote = quote_rule(prices, bid, ask)
    # クオートルールで判定できない（ミッドちょうど）部分をティックルールで埋める
    result = quote.copy()
    result[quote == 0] = tick[quote == 0]
    return result


def evaluate_accuracy(ticks: pd.DataFrame,
                      estimated: pd.Series) -> dict:
    """推定した方向と、記録された side の一致率を測る。"""
    if "side" not in ticks.columns:
        return {"error": "side列がありません"}

    actual = ticks["side"].astype(str).map({"Buy": 1, "Sell": -1})
    valid = actual.notna() & estimated.notna()
    if valid.sum() == 0:
        return {"error": "比較できるデータがありません"}

    a = actual[valid]
    e = estimated[valid]
    match = (a == e)

    # 出来高加重の一致率（大口の判定精度のほうが重要）
    if "size" in ticks.columns:
        w = ticks.loc[valid, "size"]
        weighted = float((match * w).sum() / w.sum())
    else:
        weighted = float(match.mean())

    return {
        "サンプル数": int(valid.sum()),
        "一致率": f"{match.mean():.2%}",
        "出来高加重一致率": f"{weighted:.2%}",
        "買いの再現率": f"{match[a == 1].mean():.2%}" if (a == 1).any() else "-",
        "売りの再現率": f"{match[a == -1].mean():.2%}" if (a == -1).any() else "-",
    }


def volume_imbalance(ticks: pd.DataFrame,
                     direction: pd.Series | None = None,
                     window: str = "5min") -> pd.DataFrame:
    """符号付き出来高（買い − 売り）の時系列。需給の偏りを見る。"""
    if direction is None:
        if "side" in ticks.columns:
            direction = ticks["side"].astype(str).map(
                {"Buy": 1, "Sell": -1}).fillna(0)
        else:
            direction = tick_rule(ticks["price"])

    signed = ticks["size"] * direction
    out = pd.DataFrame({
        "buy_volume": ticks["size"].where(direction > 0, 0),
        "sell_volume": ticks["size"].where(direction < 0, 0),
        "signed_volume": signed,
        "notional": ticks["price"] * ticks["size"],
    })
    agg = out.resample(window).sum()
    agg["imbalance"] = (agg["signed_volume"]
                        / (agg["buy_volume"] + agg["sell_volume"])
                        .replace(0, np.nan))
    return agg


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

    from tick_parser import load_range

    SYMBOL = "BTCUSDT"
    end = date.today() - timedelta(days=1)
    ticks = load_range(SYMBOL, end - timedelta(days=1), end)

    if ticks.empty:
        raise SystemExit("先にデータを取得してください")

    print(f"{len(ticks):,}ティックで検証\n")

    print("=== ティックルールの精度（記録されたsideと比較） ===")
    est = tick_rule(ticks["price"])
    for k, v in evaluate_accuracy(ticks, est).items():
        print(f"  {k:<18}: {v}")

    print("\n※ ティックルールの一致率は通常70〜85%程度。")
    print("  板情報を併用するLee-Readyのほうが精度は上がるが、")
    print("  Bybitのデータには side があるので推定は不要。")
    print("  この検証は『他のデータソースで推定する場合の精度の目安』になる。")

    print("\n=== 需給の偏り（5分ごと） ===")
    imb = volume_imbalance(ticks, window="5min")
    print(imb.tail(8)[["buy_volume", "sell_volume", "imbalance"]]
          .to_string(float_format=lambda v: f"{v:,.3f}"))
    print(f"\n  期間全体の偏り: {imb['imbalance'].mean():+.3f}")
    print("  （プラスなら買いが攻めている。記事09のOFIと合わせて使う）")

"""bigdata_query.py — 数十GBのティックデータをメモリに載せずに集計する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install duckdb polars
"""
from __future__ import annotations

from pathlib import Path

PARQUET_GLOB = "data/public/parquet/{symbol}/*.parquet"


# ---- DuckDB: SQLで直接Parquetを読む ------------------------------------
def duckdb_daily_stats(symbol: str = "BTCUSDT") -> None:
    import duckdb

    src = PARQUET_GLOB.format(symbol=symbol)
    q = f"""
    SELECT
        date_trunc('day', ts)                       AS day,
        count(*)                                    AS ticks,
        sum(size)                                   AS volume,
        sum(price * size)                           AS notional,
        min(price)                                  AS low,
        max(price)                                  AS high,
        arg_min(price, ts)                          AS open,
        arg_max(price, ts)                          AS close,
        sum(CASE WHEN side = 'Buy'  THEN size ELSE 0 END) AS buy_volume,
        sum(CASE WHEN side = 'Sell' THEN size ELSE 0 END) AS sell_volume
    FROM read_parquet('{src}')
    GROUP BY day
    ORDER BY day DESC
    LIMIT 15
    """
    print("=== 日次サマリ（DuckDB） ===")
    print(duckdb.sql(q))


def duckdb_hourly_liquidity(symbol: str = "BTCUSDT") -> None:
    """時間帯ごとの取引活発度。執行のタイミング設計に使う（記事19）。"""
    import duckdb

    src = PARQUET_GLOB.format(symbol=symbol)
    q = f"""
    WITH t AS (
        SELECT
            extract('hour' FROM ts)  AS hour_utc,
            size, price * size       AS notional
        FROM read_parquet('{src}')
    )
    SELECT
        hour_utc,
        (hour_utc + 9) % 24          AS hour_jst,
        count(*)                     AS ticks,
        round(sum(notional) / 1e9, 2) AS notional_bn,
        round(avg(notional), 1)      AS avg_trade_usd
    FROM t
    GROUP BY hour_utc
    ORDER BY hour_utc
    """
    print("\n=== 時間帯別の活発度（DuckDB） ===")
    print(duckdb.sql(q))


def duckdb_large_trades(symbol: str = "BTCUSDT",
                        min_notional: float = 1_000_000) -> None:
    """大口約定の抽出。マーケットインパクトの分析に使う（記事09）。"""
    import duckdb

    src = PARQUET_GLOB.format(symbol=symbol)
    q = f"""
    SELECT ts, side, price, size, round(price * size) AS notional
    FROM read_parquet('{src}')
    WHERE price * size >= {min_notional}
    ORDER BY notional DESC
    LIMIT 15
    """
    print(f"\n=== {min_notional:,.0f} USDT以上の大口約定 ===")
    print(duckdb.sql(q))


# ---- Polars: 遅延評価でストリーム処理 ----------------------------------
def polars_dollar_bars(symbol: str = "BTCUSDT",
                       threshold: float = 5_000_000) -> None:
    """Polarsの遅延評価でドルバーを作る（メモリに全部載せない）。"""
    import polars as pl

    src = PARQUET_GLOB.format(symbol=symbol)

    lf = (
        pl.scan_parquet(src)
        .select(["ts", "price", "size", "side"])
        .with_columns(
            (pl.col("price") * pl.col("size")).alias("notional"))
        .with_columns(
            # 累積売買代金を閾値で割って、バー番号を振る
            (pl.col("notional").cum_sum() / threshold)
            .floor().cast(pl.Int64).alias("bar_id"))
    )

    bars = (
        lf.group_by("bar_id")
        .agg([
            pl.col("ts").first().alias("ts_open"),
            pl.col("ts").last().alias("ts_close"),
            pl.col("price").first().alias("open"),
            pl.col("price").max().alias("high"),
            pl.col("price").min().alias("low"),
            pl.col("price").last().alias("close"),
            pl.col("size").sum().alias("volume"),
            pl.col("notional").sum().alias("dollar"),
            pl.len().alias("ticks"),
            pl.col("size").filter(pl.col("side") == "Buy")
              .sum().alias("buy_volume"),
        ])
        .sort("bar_id")
        .collect(streaming=True)      # メモリに載せずに処理
    )

    print(f"\n=== ドルバー（Polars・{threshold:,.0f} USDTごと） ===")
    print(f"  {len(bars):,}本を生成")
    print(bars.tail(5))


if __name__ == "__main__":
    SYMBOL = "BTCUSDT"
    src_dir = Path(f"data/public/parquet/{SYMBOL}")

    if not src_dir.exists() or not any(src_dir.glob("*.parquet")):
        raise SystemExit("先に public_data.py と tick_parser.py を実行してください")

    files = list(src_dir.glob("*.parquet"))
    total_mb = sum(f.stat().st_size for f in files) / 1024 / 1024
    print(f"対象: {len(files)}ファイル / {total_mb:,.0f} MB\n")

    duckdb_daily_stats(SYMBOL)
    duckdb_hourly_liquidity(SYMBOL)
    duckdb_large_trades(SYMBOL)

    try:
        polars_dollar_bars(SYMBOL)
    except ImportError:
        print("\n（polarsが未インストールのためスキップ）")

    print("\n※ DuckDBもPolarsも、Parquetを直接読むためメモリに全部載せない。")
    print("  数十GBでもノートPCで集計できる。")

"""verify_data.py — 取得したティックデータの品質を検証する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

検証:
  1) 日付の欠損（取り逃したファイルがないか）
  2) ティックからOHLCVを再構成し、APIのklineと一致するか
  3) 時刻の単調性・重複
"""
from __future__ import annotations

from dataclasses import dataclass, field
from datetime import date, timedelta
from pathlib import Path

import numpy as np
import pandas as pd

PARQUET_DIR = Path("data/public/parquet")


@dataclass
class QualityReport:
    symbol: str
    files: int = 0
    missing_days: list[str] = field(default_factory=list)
    duplicated_ts: int = 0
    non_monotonic: int = 0
    ohlcv_diff: dict = field(default_factory=dict)
    notes: list[str] = field(default_factory=list)

    @property
    def ok(self) -> bool:
        return not self.missing_days and self.non_monotonic == 0

    def report(self) -> str:
        lines = [f"{self.symbol}: {self.files}ファイル"]
        if self.missing_days:
            lines.append(f"  [!] 欠損日 {len(self.missing_days)}日: "
                         f"{self.missing_days[:5]}"
                         + (" …" if len(self.missing_days) > 5 else ""))
        else:
            lines.append("  日付の欠損: なし")
        lines.append(f"  重複タイムスタンプ: {self.duplicated_ts:,}"
                     f"（同一ミリ秒の約定は正常）")
        lines.append(f"  時刻の逆転: {self.non_monotonic}")
        if self.ohlcv_diff:
            lines.append("  APIのklineとの差:")
            for k, v in self.ohlcv_diff.items():
                lines.append(f"    {k:<8} {v}")
        for n in self.notes:
            lines.append(f"  ! {n}")
        return "\n".join(lines)


def check_missing_days(symbol: str, start: date, end: date,
                       parquet_dir: Path = PARQUET_DIR) -> list[str]:
    have = set()
    for p in (parquet_dir / symbol).glob("*.parquet"):
        try:
            have.add(date.fromisoformat(p.stem.split("_")[-1]))
        except ValueError:
            continue

    missing = []
    cur = start
    while cur <= end:
        if cur not in have:
            missing.append(str(cur))
        cur += timedelta(days=1)
    return missing


def rebuild_ohlcv(ticks: pd.DataFrame, freq: str = "1h") -> pd.DataFrame:
    """ティックからOHLCVを再構成する（APIのklineと照合するため）。"""
    return ticks.resample(freq).agg(
        open=("price", "first"), high=("price", "max"),
        low=("price", "min"), close=("price", "last"),
        volume=("size", "sum"),
    ).dropna(subset=["open"])


def compare_with_kline(rebuilt: pd.DataFrame,
                       kline: pd.DataFrame) -> dict:
    """再構成した足と、APIのklineを比較する。"""
    common = rebuilt.index.intersection(kline.index)
    if len(common) < 3:
        return {"error": "比較できる期間がありません"}

    a = rebuilt.loc[common]
    b = kline.loc[common]

    out = {}
    for col in ("open", "high", "low", "close", "volume"):
        if col not in a.columns or col not in b.columns:
            continue
        diff = (a[col] - b[col]).abs()
        rel = (diff / b[col].replace(0, np.nan)).dropna()
        out[col] = (f"最大 {diff.max():.6f} / "
                    f"相対 {rel.max():.4%} / 一致 {(rel < 1e-6).mean():.1%}")
    out["比較本数"] = len(common)
    return out


def verify(symbol: str, days: int = 7,
           http=None) -> QualityReport:
    from tick_parser import load_range

    end = date.today() - timedelta(days=1)
    start = end - timedelta(days=days - 1)

    rep = QualityReport(symbol=symbol)
    rep.files = len(list((PARQUET_DIR / symbol).glob("*.parquet")))
    rep.missing_days = check_missing_days(symbol, start, end)

    ticks = load_range(symbol, start, end)
    if ticks.empty:
        rep.notes.append("データが読み込めません")
        return rep

    rep.duplicated_ts = int(ticks.index.duplicated().sum())
    rep.non_monotonic = int((ticks.index.to_series().diff()
                             < pd.Timedelta(0)).sum())

    if http is not None:
        from kline_fetch import fetch_klines

        kl = fetch_klines(http, symbol, interval="60",
                          start=start.isoformat(), verbose=False)
        if not kl.empty:
            rebuilt = rebuild_ohlcv(ticks, "1h")
            rep.ohlcv_diff = compare_with_kline(rebuilt, kl)

    return rep


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

    c = make_client()
    rep = verify("BTCUSDT", days=7, http=c.http)
    print(rep.report())

    if rep.ok:
        print("\nデータ品質は良好です。バックテストに使用できます。")
    else:
        print("\n[!] 問題があります。欠損日を再ダウンロードしてください。")

    print("\n※ ティックから再構成したOHLCVがAPIのklineとほぼ一致すれば、")
    print("  データの正確性が確認できる。乖離が大きい場合は、")
    print("  タイムゾーンや足の境界（記事07）の扱いを確認すること。")