From 52d05f2ddd99f34ef988593ba56112aca8e79c4d Mon Sep 17 00:00:00 2001 From: 21in7 Date: Mon, 4 May 2026 09:03:06 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=EC=A0=84=EB=9E=B5=20=EB=A6=AC=EC=84=9C?= =?UTF-8?q?=EC=B9=98=20=EC=8A=A4=ED=81=AC=EB=A6=BD=ED=8A=B8=20=EB=B0=8F=20?= =?UTF-8?q?=ED=85=8C=EC=8A=A4=ED=8A=B8=20=EC=9D=BC=EA=B4=84=20=EC=B6=94?= =?UTF-8?q?=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - FR/OI 백테스트, LS ratio 백테스트 스크립트 - 펀딩/OI 분석, 거래 LS 분석 스크립트 - evaluate_oos 테스트 추가 Co-Authored-By: Claude Opus 4.6 (1M context) --- scripts/fr_oi_backtest.py | 364 +++++++++++++++++++++++++ scripts/funding_oi_analysis.py | 312 +++++++++++++++++++++ scripts/ls_ratio_backtest.py | 485 +++++++++++++++++++++++++++++++++ scripts/revalidate_apr15.py | 49 ++++ scripts/trade_ls_analysis.py | 459 +++++++++++++++++++++++++++++++ tests/test_evaluate_oos.py | 181 ++++++++++++ 6 files changed, 1850 insertions(+) create mode 100644 scripts/fr_oi_backtest.py create mode 100644 scripts/funding_oi_analysis.py create mode 100644 scripts/ls_ratio_backtest.py create mode 100644 scripts/revalidate_apr15.py create mode 100644 scripts/trade_ls_analysis.py create mode 100644 tests/test_evaluate_oos.py diff --git a/scripts/fr_oi_backtest.py b/scripts/fr_oi_backtest.py new file mode 100644 index 0000000..5960250 --- /dev/null +++ b/scripts/fr_oi_backtest.py @@ -0,0 +1,364 @@ +""" +FR × OI 변화율 백테스트 — Phase 1: 12개 조합 + +신호: FR × OI변화율(1h) = funding_rate × oi_pct_change_4 +- SHORT: 피처 >= threshold (롱 스퀴즈 전조) +- LONG: 피처 <= threshold (숏 스퀴즈 전조) +- 보유: 1h(4캔들) / 4h(16캔들) + +Usage: python scripts/fr_oi_backtest.py +""" + +import asyncio +import aiohttp +import pandas as pd +import numpy as np +from datetime import datetime, timedelta, timezone +from pathlib import Path + +BASE = "https://fapi.binance.com" +SYMBOL = "XRPUSDT" +DATA_DIR = Path("data/xrpusdt") +FEE_RATE = 0.0004 + + +async def fetch_oi_history(session, symbol, start_ms, end_ms): + all_data = [] + current = start_ms + calls = 0 + while current < end_ms: + params = {"symbol": symbol, "period": "15m", "startTime": current, "endTime": end_ms, "limit": 500} + async with session.get(f"{BASE}/futures/data/openInterestHist", params=params) as resp: + data = await resp.json() + if not data or not isinstance(data, list): + break + all_data.extend(data) + last_ts = int(data[-1]["timestamp"]) + if last_ts <= current: + break + current = last_ts + 1 + calls += 1 + if calls % 50 == 0: + await asyncio.sleep(5) + else: + await asyncio.sleep(0.1) + if not all_data: + return pd.DataFrame() + df = pd.DataFrame(all_data) + df["timestamp"] = pd.to_datetime(df["timestamp"].astype(int), unit="ms", utc=True) + df["oi_value"] = df["sumOpenInterestValue"].astype(float) + return df[["timestamp", "oi_value"]].drop_duplicates("timestamp").sort_values("timestamp") + + +async def fetch_funding_rate(session, symbol, start_ms, end_ms): + all_data = [] + current = start_ms + while current < end_ms: + params = {"symbol": symbol, "startTime": current, "endTime": end_ms, "limit": 1000} + async with session.get(f"{BASE}/fapi/v1/fundingRate", params=params) as resp: + data = await resp.json() + if not data or not isinstance(data, list): + break + all_data.extend(data) + last_ts = int(data[-1]["fundingTime"]) + if last_ts <= current: + break + current = last_ts + 1 + await asyncio.sleep(0.1) + if not all_data: + return pd.DataFrame() + df = pd.DataFrame(all_data) + df["timestamp"] = pd.to_datetime(df["fundingTime"].astype(int), unit="ms", utc=True) + df["funding_rate"] = df["fundingRate"].astype(float) + return df[["timestamp", "funding_rate"]].drop_duplicates("timestamp").sort_values("timestamp") + + +def run_backtest(df, feature_col, percentile, direction, hold_bars): + threshold = df[feature_col].quantile(percentile / 100) + trades = [] + i = 0 + while i < len(df) - hold_bars - 1: + val = df.iloc[i][feature_col] + if pd.isna(val): + i += 1 + continue + + trigger = False + if direction == "SHORT" and val >= threshold: + trigger = True + elif direction == "LONG" and val <= threshold: + trigger = True + + if trigger: + entry_idx = i + 1 + exit_idx = i + 1 + hold_bars - 1 + if exit_idx >= len(df): + break + entry_price = df.iloc[entry_idx]["open"] + exit_price = df.iloc[exit_idx]["close"] + + if direction == "LONG": + gross_return = (exit_price / entry_price) - 1 + else: + gross_return = (entry_price / exit_price) - 1 + + fee = FEE_RATE * 2 + net_return = gross_return - fee + + trades.append({ + "entry_time": df.iloc[entry_idx]["timestamp"], + "exit_time": df.iloc[exit_idx]["timestamp"], + "entry_price": entry_price, + "exit_price": exit_price, + "feature_val": val, + "gross_return_bps": gross_return * 10000, + "net_return_bps": net_return * 10000, + }) + i = exit_idx + 1 # 포지션 종료 후 다음 + else: + i += 1 + + if not trades: + return None + + tdf = pd.DataFrame(trades) + wins = tdf[tdf["net_return_bps"] > 0]["net_return_bps"] + losses = tdf[tdf["net_return_bps"] <= 0]["net_return_bps"] + + gross_profit = wins.sum() if len(wins) > 0 else 0 + gross_loss = abs(losses.sum()) if len(losses) > 0 else 0 + pf = gross_profit / gross_loss if gross_loss > 0 else float("inf") if gross_profit > 0 else 0 + + cum_pnl = tdf["net_return_bps"].cumsum() + max_dd = (cum_pnl - cum_pnl.cummax()).min() + + return { + "trades": len(tdf), + "wins": len(wins), + "losses": len(losses), + "win_rate": len(wins) / len(tdf) * 100, + "pf": pf, + "total_pnl_bps": tdf["net_return_bps"].sum(), + "avg_pnl_bps": tdf["net_return_bps"].mean(), + "max_dd_bps": max_dd, + "threshold": threshold, + "df_trades": tdf, + } + + +def confidence(n): + if n < 20: + return "🔴", "폐기" + elif n < 50: + return "🟡", "참고" + else: + return "🟢", "검토" + + +async def main(): + print("=" * 80) + print(" FR × OI 변화율 백테스트 — Phase 1: 12개 조합") + print("=" * 80) + + # 데이터 수집 + print("\n[1] 데이터 수집") + df_kline = pd.read_parquet(DATA_DIR / "combined_15m.parquet") + + end_dt = datetime.now(timezone.utc) + oi_start_dt = end_dt - timedelta(days=29) + oi_start_ms = int(oi_start_dt.replace(microsecond=0, second=0).timestamp()) * 1000 + fr_start_ms = oi_start_ms + end_ms = int(end_dt.replace(microsecond=0, second=0).timestamp()) * 1000 + + async with aiohttp.ClientSession() as session: + print(" OI 수집...") + oi_df = await fetch_oi_history(session, SYMBOL, oi_start_ms, end_ms) + print(f" OI: {len(oi_df)} rows") + print(" FR 수집...") + fr_df = await fetch_funding_rate(session, SYMBOL, fr_start_ms, end_ms) + print(f" FR: {len(fr_df)} rows") + + # 병합 + print("\n[2] 데이터 병합") + df = df_kline.loc[oi_start_dt:].copy().reset_index() + print(f" Kline (29일): {len(df)} rows") + + # OI 병합 + df = pd.merge_asof(df.sort_values("timestamp"), oi_df.sort_values("timestamp"), + on="timestamp", direction="nearest", tolerance=pd.Timedelta(minutes=20)) + df["oi_pct_change_4"] = df["oi_value"].pct_change(4) + + # FR 병합 (forward fill) + df = pd.merge_asof(df.sort_values("timestamp"), fr_df.rename(columns={"funding_rate": "fr_api"}).sort_values("timestamp"), + on="timestamp", direction="backward") + + # 핵심 피처: FR × OI변화율(1h) + df["fr_x_oi_1h"] = df["fr_api"] * df["oi_pct_change_4"] + + valid = df.dropna(subset=["fr_x_oi_1h"]) + print(f" 유효 데이터: {len(valid)} rows") + print(f" fr_x_oi_1h: mean={valid['fr_x_oi_1h'].mean():.8f}, std={valid['fr_x_oi_1h'].std():.8f}") + + for p in [25, 50, 75]: + v = valid["fr_x_oi_1h"].quantile(p / 100) + print(f" P{p}: {v:.8f}") + + # 12개 조합 백테스트 + print("\n[3] 12개 조합 백테스트") + print("=" * 80) + + combos = [] + for hold_label, hold_bars in [("1h", 4), ("4h", 16)]: + for direction in ["SHORT", "LONG"]: + for pct in [75, 50, 25]: + desc_dir = "롱스퀴즈" if direction == "SHORT" else "숏스퀴즈" + combos.append({ + "hold_label": hold_label, + "hold_bars": hold_bars, + "direction": direction, + "percentile": pct, + "desc": f"{direction} {hold_label} P{pct} ({desc_dir})", + }) + + results = [] + for c in combos: + r = run_backtest(valid.reset_index(drop=True), "fr_x_oi_1h", + c["percentile"], c["direction"], c["hold_bars"]) + if r: + r.update(c) + else: + r = {**c, "trades": 0, "wins": 0, "losses": 0, "win_rate": 0, + "pf": 0, "total_pnl_bps": 0, "avg_pnl_bps": 0, "max_dd_bps": 0, "threshold": 0} + results.append(r) + + # 결과 테이블 + print(f"\n{'ID':>3} {'조합':<28} {'거래수':>6} {'승률':>7} {'PF':>7} {'PnL(bps)':>10} {'MaxDD':>10} {'신뢰도'}") + print("-" * 90) + + for i, r in enumerate(results, 1): + emoji, label = confidence(r["trades"]) + pf_str = f"{r['pf']:.2f}" if r["pf"] != float("inf") else "INF" + print(f"{i:>3} {r['desc']:<28} {r['trades']:>6} {r['win_rate']:>6.1f}% {pf_str:>7} " + f"{r['total_pnl_bps']:>+10.1f} {r['max_dd_bps']:>10.1f} {emoji} {label}") + + # 대칭성 검증 + print("\n" + "=" * 80) + print(" [대칭성 검증]") + print("=" * 80) + + for hold_label in ["1h", "4h"]: + shorts = [r for r in results if r["hold_label"] == hold_label and r["direction"] == "SHORT" and r["trades"] > 0] + longs = [r for r in results if r["hold_label"] == hold_label and r["direction"] == "LONG" and r["trades"] > 0] + + best_short = max(shorts, key=lambda x: x["pf"]) if shorts else None + best_long = max(longs, key=lambda x: x["pf"]) if longs else None + + print(f"\n [{hold_label} 보유]") + if best_short: + print(f" Best SHORT: {best_short['desc']} — PF={best_short['pf']:.2f}, {best_short['trades']}건") + if best_long: + print(f" Best LONG: {best_long['desc']} — PF={best_long['pf']:.2f}, {best_long['trades']}건") + + if best_short and best_long: + s_pf = best_short["pf"] + l_pf = best_long["pf"] + if s_pf > 1.5 and l_pf > 1.5: + print(f" → Case 1: 양방향 생존 ✓ Phase 2 후보") + elif (s_pf > 1.5 and l_pf < 0.5) or (l_pf > 1.5 and s_pf < 0.5): + print(f" → Case 2: 한쪽만 성공 ✗ 시장 베타/우연") + elif s_pf > 1.5 or l_pf > 1.5: + print(f" → Case 3: 부분적 edge ~ 낮은 신뢰도") + elif s_pf > 1.0 and l_pf > 1.0: + print(f" → 양쪽 PF > 1.0이나 < 1.5 — 약한 edge") + else: + print(f" → 양쪽 모두 약함") + + # 보유시간 비교 + print("\n" + "=" * 80) + print(" [보유시간 비교]") + print("=" * 80) + + for direction in ["SHORT", "LONG"]: + r_1h = [r for r in results if r["hold_label"] == "1h" and r["direction"] == direction and r["trades"] > 0] + r_4h = [r for r in results if r["hold_label"] == "4h" and r["direction"] == direction and r["trades"] > 0] + best_1h = max(r_1h, key=lambda x: x["pf"]) if r_1h else None + best_4h = max(r_4h, key=lambda x: x["pf"]) if r_4h else None + + print(f"\n [{direction}]") + if best_1h: + print(f" 1h Best: PF={best_1h['pf']:.2f} ({best_1h['desc']}, {best_1h['trades']}건)") + if best_4h: + print(f" 4h Best: PF={best_4h['pf']:.2f} ({best_4h['desc']}, {best_4h['trades']}건)") + if best_1h and best_4h: + if best_4h["pf"] > best_1h["pf"]: + print(f" → 4h가 더 강함 (상관분석 r=-0.1734과 일치)") + else: + print(f" → 1h가 더 강함 (주의: 상관분석은 4h 기준)") + + # 최종 판정 + print("\n" + "=" * 80) + print(" [최종 판정]") + print("=" * 80) + + # Phase 2 후보 찾기 + phase2 = [] + for hold_label in ["4h", "1h"]: + shorts = [r for r in results if r["hold_label"] == hold_label and r["direction"] == "SHORT" and r["trades"] >= 20] + longs = [r for r in results if r["hold_label"] == hold_label and r["direction"] == "LONG" and r["trades"] >= 20] + + best_s = max(shorts, key=lambda x: x["pf"]) if shorts else None + best_l = max(longs, key=lambda x: x["pf"]) if longs else None + + if best_s and best_l: + if best_s["pf"] > 1.5 and best_l["pf"] > 1.5: + phase2.append(("Case1", hold_label, best_s, best_l)) + elif best_s["pf"] > 1.5 or best_l["pf"] > 1.5: + phase2.append(("Case3", hold_label, best_s, best_l)) + + if phase2: + print(f"\n 🟢 Phase 2 후보 발견!") + for case, hl, bs, bl in phase2: + print(f" [{case}] {hl}: SHORT PF={bs['pf']:.2f}({bs['trades']}건), " + f"LONG PF={bl['pf']:.2f}({bl['trades']}건)") + print(f"\n → Phase 2 (Bot Simulation) 진행 권장") + print(f" → 단, 29일 OI 데이터 + 448행 제한 감안") + else: + all_pf = [(r["desc"], r["pf"], r["trades"]) for r in results if r["trades"] > 0] + all_pf.sort(key=lambda x: x[1], reverse=True) + best = all_pf[0] if all_pf else ("N/A", 0, 0) + + above_1 = [r for r in results if r["pf"] > 1.0 and r["trades"] >= 20] + if above_1: + print(f"\n 🟡 PF > 1.0 조합 존재 ({len(above_1)}개), 단 < 1.5") + for r in sorted(above_1, key=lambda x: x["pf"], reverse=True): + emoji, _ = confidence(r["trades"]) + print(f" {r['desc']}: PF={r['pf']:.2f}, {r['trades']}건 {emoji}") + print(f"\n → 약한 edge. 4월 데이터 축적 후 재검증 권장.") + else: + print(f"\n 🔴 PF > 1.0 조합 없음 (20건 이상)") + print(f" Best: {best[0]} (PF={best[1]:.2f}, {best[2]}건)") + print(f"\n → FR × OI 시그널도 비용 후 edge 없음") + + # Best 조합 상세 + valid_results = [r for r in results if r["trades"] > 10 and "df_trades" in r] + if valid_results: + best_r = max(valid_results, key=lambda x: x["pf"]) + print(f"\n[참고] Best 조합 상세: {best_r['desc']}") + print("-" * 60) + tdf = best_r["df_trades"] + print(f" 기간: {tdf['entry_time'].min()} ~ {tdf['exit_time'].max()}") + print(f" 평균 피처값: {tdf['feature_val'].mean():.8f}") + w = tdf[tdf["net_return_bps"] > 0] + l = tdf[tdf["net_return_bps"] <= 0] + if len(w) > 0: + print(f" 수익 거래 평균: {w['net_return_bps'].mean():.1f} bps ({len(w)}건)") + if len(l) > 0: + print(f" 손실 거래 평균: {l['net_return_bps'].mean():.1f} bps ({len(l)}건)") + + print("\n" + "=" * 80) + print(" 분석 완료.") + print("=" * 80) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/scripts/funding_oi_analysis.py b/scripts/funding_oi_analysis.py new file mode 100644 index 0000000..5dc908b --- /dev/null +++ b/scripts/funding_oi_analysis.py @@ -0,0 +1,312 @@ +""" +Funding Rate + OI 변화율 상관분석 + +기존 combined_15m.parquet에 funding_rate 2년치 있음. +OI는 Binance API에서 2개월치 수집 후 병합. +상관분석 → r 값으로 edge 판정. + +Usage: python scripts/funding_oi_analysis.py +""" + +import asyncio +import aiohttp +import pandas as pd +import numpy as np +from datetime import datetime, timedelta, timezone +from pathlib import Path +import time + +BASE = "https://fapi.binance.com" +SYMBOL = "XRPUSDT" +DATA_DIR = Path("data/xrpusdt") +FEE_RATE = 0.0004 # 0.04% per side + + +async def fetch_oi_history(session, symbol, start_ms, end_ms): + """Binance Open Interest Statistics (15m) 수집""" + all_data = [] + current = start_ms + calls = 0 + + while current < end_ms: + params = { + "symbol": symbol, + "period": "15m", + "startTime": current, + "endTime": end_ms, + "limit": 500, + } + async with session.get(f"{BASE}/futures/data/openInterestHist", params=params) as resp: + data = await resp.json() + + if not data or not isinstance(data, list): + break + + all_data.extend(data) + last_ts = int(data[-1]["timestamp"]) + if last_ts <= current: + break + current = last_ts + 1 + calls += 1 + + # Rate limit: ~10 weight per call, 1200/min limit + if calls % 50 == 0: + print(f" ... {len(all_data)} rows fetched, sleeping 5s for rate limit") + await asyncio.sleep(5) + else: + await asyncio.sleep(0.1) + + if not all_data: + return pd.DataFrame() + + df = pd.DataFrame(all_data) + df["timestamp"] = pd.to_datetime(df["timestamp"].astype(int), unit="ms", utc=True) + df["sumOpenInterest"] = df["sumOpenInterest"].astype(float) + df["sumOpenInterestValue"] = df["sumOpenInterestValue"].astype(float) + return df[["timestamp", "sumOpenInterest", "sumOpenInterestValue"]].drop_duplicates("timestamp").sort_values("timestamp") + + +async def fetch_funding_rate_history(session, symbol, start_ms, end_ms): + """Binance Funding Rate History 수집 (8시간 간격)""" + all_data = [] + current = start_ms + + while current < end_ms: + params = { + "symbol": symbol, + "startTime": current, + "endTime": end_ms, + "limit": 1000, + } + async with session.get(f"{BASE}/fapi/v1/fundingRate", params=params) as resp: + data = await resp.json() + + if not data or not isinstance(data, list): + break + + all_data.extend(data) + last_ts = int(data[-1]["fundingTime"]) + if last_ts <= current: + break + current = last_ts + 1 + await asyncio.sleep(0.1) + + if not all_data: + return pd.DataFrame() + + df = pd.DataFrame(all_data) + df["timestamp"] = pd.to_datetime(df["fundingTime"].astype(int), unit="ms", utc=True) + df["funding_rate_api"] = df["fundingRate"].astype(float) + return df[["timestamp", "funding_rate_api"]].drop_duplicates("timestamp").sort_values("timestamp") + + +async def main(): + print("=" * 80) + print(" Funding Rate + OI 변화율 상관분석") + print("=" * 80) + + # Step 1: 데이터 수집 + print("\n[Step 1] 데이터 수집") + + # 기존 kline 로드 + kline_path = DATA_DIR / "combined_15m.parquet" + df = pd.read_parquet(kline_path) + print(f" 기존 kline: {len(df)} rows ({df.index.min()} ~ {df.index.max()})") + + # 기간 설정: OI는 30일 제한, FR은 무제한 + end_dt = datetime.now(timezone.utc) + oi_start_dt = end_dt - timedelta(days=29) # OI: 30일 제한 + fr_start_dt = end_dt - timedelta(days=60) # FR: 60일 + kline_start_dt = fr_start_dt # kline도 60일 + + # Clean timestamps (no microseconds) + oi_start_ms = int(oi_start_dt.replace(microsecond=0, second=0).timestamp()) * 1000 + fr_start_ms = int(fr_start_dt.replace(microsecond=0, second=0).timestamp()) * 1000 + end_ms = int(end_dt.replace(microsecond=0, second=0).timestamp()) * 1000 + + print(f" OI 수집 기간: {oi_start_dt.date()} ~ {end_dt.date()} (29일)") + print(f" FR 수집 기간: {fr_start_dt.date()} ~ {end_dt.date()} (60일)") + + async with aiohttp.ClientSession() as session: + print(" OI 수집 중...") + oi_df = await fetch_oi_history(session, SYMBOL, oi_start_ms, end_ms) + print(f" OI: {len(oi_df)} rows") + + print(" Funding Rate 수집 중...") + fr_df = await fetch_funding_rate_history(session, SYMBOL, fr_start_ms, end_ms) + print(f" Funding Rate: {len(fr_df)} rows") + + # Step 2: 병합 + print("\n[Step 2] 데이터 병합") + + # 2개월 kline 슬라이스 + df_2m = df.loc[kline_start_dt:].copy() + print(f" 2개월 kline: {len(df_2m)} rows") + + # OI 병합 (merge_asof) + df_2m = df_2m.reset_index() + if not oi_df.empty: + df_2m = pd.merge_asof( + df_2m.sort_values("timestamp"), + oi_df.sort_values("timestamp"), + on="timestamp", + direction="nearest", + tolerance=pd.Timedelta(minutes=20), + ) + # OI 변화율 계산 + df_2m["oi"] = df_2m["sumOpenInterestValue"] + df_2m["oi_pct_change"] = df_2m["oi"].pct_change() + df_2m["oi_pct_change_4"] = df_2m["oi"].pct_change(4) # 1시간 변화율 + print(f" OI 매칭: {df_2m['oi'].notna().sum()} rows") + + # Funding Rate 병합 (8h → 15m forward fill) + if not fr_df.empty: + df_2m = pd.merge_asof( + df_2m.sort_values("timestamp"), + fr_df.sort_values("timestamp"), + on="timestamp", + direction="backward", # 가장 최근 funding rate 사용 + ) + # Funding rate 변화율 + df_2m["fr"] = df_2m["funding_rate_api"] + df_2m["fr_change"] = df_2m["fr"].diff() + print(f" Funding Rate 매칭: {df_2m['fr'].notna().sum()} rows") + + # 기존 funding_rate 컬럼도 활용 + df_2m["fr_existing"] = df_2m["funding_rate"] + df_2m["fr_existing_change"] = df_2m["fr_existing"].diff() + + # 미래 수익률 계산 + df_2m["next_1h_return"] = df_2m["close"].shift(-4) / df_2m["close"] - 1 + df_2m["next_4h_return"] = df_2m["close"].shift(-16) / df_2m["close"] - 1 + df_2m["next_15m_return"] = df_2m["close"].shift(-1) / df_2m["close"] - 1 + + # 복합 피처 + if "oi_pct_change" in df_2m.columns and "fr" in df_2m.columns: + df_2m["fr_x_oi"] = df_2m["fr"] * df_2m["oi_pct_change"] # 펀딩비 × OI변화율 + df_2m["fr_x_oi_4"] = df_2m["fr"] * df_2m["oi_pct_change_4"] + + df_2m = df_2m.set_index("timestamp") + + # OI velocity (변화율의 변화율) + if "oi_pct_change" in df_2m.columns: + df_2m["oi_velocity"] = df_2m["oi_pct_change"].diff() + df_2m["oi_acceleration"] = df_2m["oi_velocity"].diff() + + print(f"\n 최종 데이터셋: {len(df_2m)} rows, {len(df_2m.columns)} columns") + + # Step 3: 상관분석 + print("\n[Step 3] 상관분석") + print("=" * 80) + + features = [ + ("fr_existing", "Funding Rate (기존)"), + ("fr_existing_change", "ΔFunding Rate"), + ("fr", "Funding Rate (API)"), + ("fr_change", "ΔFunding Rate (API)"), + ("oi_pct_change", "OI 변화율 (15m)"), + ("oi_pct_change_4", "OI 변화율 (1h)"), + ("oi_velocity", "OI Velocity"), + ("oi_acceleration", "OI Acceleration"), + ("fr_x_oi", "FR × OI변화율"), + ("fr_x_oi_4", "FR × OI변화율(1h)"), + ] + + targets = [ + ("next_15m_return", "다음 15m"), + ("next_1h_return", "다음 1h"), + ("next_4h_return", "다음 4h"), + ] + + print(f"\n{'피처':<25} {'→15m':>8} {'→1h':>8} {'→4h':>8} {'N':>7}") + print("-" * 60) + + strong_signals = [] + for feat_col, feat_name in features: + if feat_col not in df_2m.columns: + continue + corrs = [] + n = 0 + for tgt_col, _ in targets: + valid = df_2m[[feat_col, tgt_col]].dropna() + n = len(valid) + if n > 50: + r = valid[feat_col].corr(valid[tgt_col]) + corrs.append(r) + else: + corrs.append(float("nan")) + + r_strs = [f"{r:>+8.4f}" if not np.isnan(r) else f"{'N/A':>8}" for r in corrs] + print(f"{feat_name:<25} {''.join(r_strs)} {n:>7}") + + # 강한 시그널 체크 (|r| > 0.05) + for r, (tgt_col, tgt_name) in zip(corrs, targets): + if not np.isnan(r) and abs(r) > 0.05: + strong_signals.append((feat_name, tgt_name, r, n)) + + # Quintile 분석 (강한 시그널에 대해) + print("\n" + "=" * 80) + print(" [Quintile 분석] |r| > 0.05 피처") + print("=" * 80) + + for feat_col, feat_name in features: + if feat_col not in df_2m.columns: + continue + + for tgt_col, tgt_name in targets: + valid = df_2m[[feat_col, tgt_col]].dropna() + if len(valid) < 100: + continue + r = valid[feat_col].corr(valid[tgt_col]) + if abs(r) < 0.05: + continue + + print(f"\n {feat_name} → {tgt_name} (r={r:+.4f}, n={len(valid)})") + print(f" {'Quintile':<12} {'mean_feat':>12} {'return_bps':>12} {'win_rate':>10} {'count':>7}") + print(" " + "-" * 55) + + try: + valid["q"] = pd.qcut(valid[feat_col], 5, labels=["Q1", "Q2", "Q3", "Q4", "Q5"], duplicates="drop") + except ValueError: + continue + + for q in valid["q"].cat.categories: + grp = valid[valid["q"] == q] + if len(grp) == 0: + continue + mr = grp[feat_col].mean() + ret = grp[tgt_col].mean() * 10000 + wr = (grp[tgt_col] > 0).mean() * 100 + print(f" {q:<12} {mr:>12.6f} {ret:>+12.2f} {wr:>9.1f}% {len(grp):>7}") + + # 판정 + print("\n" + "=" * 80) + print(" [최종 판정]") + print("=" * 80) + + if strong_signals: + print(f"\n |r| > 0.05 시그널: {len(strong_signals)}개") + for feat, tgt, r, n in sorted(strong_signals, key=lambda x: abs(x[2]), reverse=True): + marker = "🟢" if abs(r) > 0.15 else "🟡" if abs(r) > 0.10 else "⚪" + print(f" {marker} {feat} → {tgt}: r={r:+.4f} (n={n})") + + best_r = max(abs(r) for _, _, r, _ in strong_signals) + if best_r > 0.15: + print(f"\n ✅ r > 0.15 시그널 발견! 백테스트 진행 가치 있음") + elif best_r > 0.10: + print(f"\n 🟡 r = 0.10~0.15. L/S ratio(0.1158)과 비슷한 수준.") + print(f" 단, 2개월 데이터(8일 대비 7.5배)이므로 신뢰도 높음.") + print(f" 백테스트로 비용 후 PF 확인 필요.") + else: + print(f"\n ⚠️ 최대 |r| = {best_r:.4f}. 약한 시그널.") + print(f" 비용(0.08%) 커버 가능성 낮음.") + else: + print("\n 🔴 |r| > 0.05 시그널 없음. Edge 없음.") + + print("\n" + "=" * 80) + print(" 분석 완료.") + print("=" * 80) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/scripts/ls_ratio_backtest.py b/scripts/ls_ratio_backtest.py new file mode 100644 index 0000000..58e3f9d --- /dev/null +++ b/scripts/ls_ratio_backtest.py @@ -0,0 +1,485 @@ +""" +L/S Ratio 단독 백테스트 — Phase 1: Pure Edge Test + +6개 조합 (3 임계값 × 2 방향) 스윕, 3단계 필터 판정. +데이터: 프로덕션 수집 L/S ratio + Binance kline (같은 기간). + +Usage: python scripts/ls_ratio_backtest.py +""" + +import asyncio +import aiohttp +import pandas as pd +import numpy as np +from datetime import timezone +from pathlib import Path + +BASE = "https://fapi.binance.com" +DATA_DIR = Path("data") +SYMBOL = "XRPUSDT" +FEE_RATE = 0.0004 # 0.04% per side +HOLD_BARS = 4 # 4 candles = 1 hour + + +async def fetch_klines(session, symbol, start_ms, end_ms): + """Binance kline 데이터 가져오기""" + all_klines = [] + current = start_ms + while current < end_ms: + params = { + "symbol": symbol, "interval": "15m", + "startTime": current, "endTime": end_ms, "limit": 1500, + } + async with session.get(f"{BASE}/fapi/v1/klines", params=params) as resp: + data = await resp.json() + if not data: + break + all_klines.extend(data) + current = data[-1][0] + 1 + return all_klines + + +def load_ls_ratio(symbol): + """프로덕션 수집 L/S ratio 로드""" + path = DATA_DIR / symbol.lower() / "ls_ratio_15m.parquet" + if not path.exists(): + raise FileNotFoundError(f"{path} not found. Sync from production first.") + df = pd.read_parquet(path) + df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True) + return df.sort_values("timestamp").reset_index(drop=True) + + +def build_dataset(klines_raw, ls_df): + """Kline + L/S ratio 조인""" + df = pd.DataFrame(klines_raw, columns=[ + "open_time", "open", "high", "low", "close", "volume", + "close_time", "quote_vol", "trades", "taker_buy_vol", + "taker_buy_quote_vol", "ignore", + ]) + df["timestamp"] = pd.to_datetime(df["open_time"], unit="ms", utc=True) + for c in ["open", "high", "low", "close", "volume"]: + df[c] = df[c].astype(float) + + # L/S ratio 조인 (가장 가까운 타임스탬프) + df = df.sort_values("timestamp").reset_index(drop=True) + merged = pd.merge_asof( + df, ls_df, on="timestamp", direction="nearest", + tolerance=pd.Timedelta(minutes=20), + ) + return merged + + +def run_backtest(df, percentile, direction, hold_bars=HOLD_BARS): + """ + 단일 조합 백테스트 실행. + + - percentile: L/S ratio 임계값 (0~100) + - direction: "LONG" or "SHORT" + - hold_bars: 보유 캔들 수 + + LONG 진입: ratio >= threshold (Momentum) + SHORT 진입: ratio <= threshold (Momentum) + """ + threshold = df["top_acct_ls_ratio"].quantile(percentile / 100) + + trades = [] + i = 0 + while i < len(df) - hold_bars: + ratio = df.iloc[i]["top_acct_ls_ratio"] + if pd.isna(ratio): + i += 1 + continue + + # 시그널 체크 + if direction == "LONG" and ratio >= threshold: + entry_price = df.iloc[i + 1]["open"] # 다음 캔들 시가 진입 + exit_price = df.iloc[i + 1 + hold_bars - 1]["close"] # hold_bars 후 종가 + gross_return = (exit_price / entry_price) - 1 + fee = FEE_RATE * 2 # 진입 + 청산 + net_return = gross_return - fee + trades.append({ + "entry_time": df.iloc[i + 1]["timestamp"], + "exit_time": df.iloc[i + 1 + hold_bars - 1]["timestamp"], + "entry_price": entry_price, + "exit_price": exit_price, + "entry_ls_ratio": ratio, + "gross_return_bps": gross_return * 10000, + "net_return_bps": net_return * 10000, + "fee_bps": fee * 10000, + }) + i += 1 + hold_bars # 포지션 종료 후 다음 캔들부터 + elif direction == "SHORT" and ratio <= threshold: + entry_price = df.iloc[i + 1]["open"] + exit_price = df.iloc[i + 1 + hold_bars - 1]["close"] + gross_return = (entry_price / exit_price) - 1 # SHORT: 반대 + fee = FEE_RATE * 2 + net_return = gross_return - fee + trades.append({ + "entry_time": df.iloc[i + 1]["timestamp"], + "exit_time": df.iloc[i + 1 + hold_bars - 1]["timestamp"], + "entry_price": entry_price, + "exit_price": exit_price, + "entry_ls_ratio": ratio, + "gross_return_bps": gross_return * 10000, + "net_return_bps": net_return * 10000, + "fee_bps": fee * 10000, + }) + i += 1 + hold_bars + else: + i += 1 + + if not trades: + return None + + df_trades = pd.DataFrame(trades) + + # PF 계산: Σ(net profit) / Σ(|net loss|) + wins = df_trades[df_trades["net_return_bps"] > 0]["net_return_bps"] + losses = df_trades[df_trades["net_return_bps"] <= 0]["net_return_bps"] + + gross_profit = wins.sum() if len(wins) > 0 else 0 + gross_loss = abs(losses.sum()) if len(losses) > 0 else 0 + + pf = gross_profit / gross_loss if gross_loss > 0 else float("inf") if gross_profit > 0 else 0 + + # Max Drawdown (cumulative bps) + cum_pnl = df_trades["net_return_bps"].cumsum() + running_max = cum_pnl.cummax() + drawdown = cum_pnl - running_max + max_dd = drawdown.min() + + return { + "trades": len(df_trades), + "wins": len(wins), + "losses": len(losses), + "win_rate": len(wins) / len(df_trades) * 100, + "pf": pf, + "total_pnl_bps": df_trades["net_return_bps"].sum(), + "avg_pnl_bps": df_trades["net_return_bps"].mean(), + "max_dd_bps": max_dd, + "threshold": threshold, + "df_trades": df_trades, + } + + +def confidence_emoji(n_trades): + if n_trades < 20: + return "🔴" + elif n_trades < 50: + return "🟡" + elif n_trades < 100: + return "🟢" + else: + return "🟢" + + +def confidence_label(n_trades): + if n_trades < 20: + return "폐기(과적합)" + elif n_trades < 50: + return "낮음(참고만)" + elif n_trades < 100: + return "보통(검토)" + else: + return "높음(우선)" + + +async def main(): + print("=" * 80) + print(" L/S Ratio 단독 백테스트 — Phase 1: Pure Edge Test") + print("=" * 80) + + # 1. 데이터 로드 + print("\n[1] 데이터 로드") + ls_df = load_ls_ratio(SYMBOL) + print(f" L/S ratio: {len(ls_df)} rows ({ls_df['timestamp'].min()} ~ {ls_df['timestamp'].max()})") + + start_ms = int(ls_df["timestamp"].min().timestamp() * 1000) + end_ms = int(ls_df["timestamp"].max().timestamp() * 1000) + + async with aiohttp.ClientSession() as session: + klines = await fetch_klines(session, SYMBOL, start_ms, end_ms) + print(f" Klines: {len(klines)} rows") + + df = build_dataset(klines, ls_df) + valid = df.dropna(subset=["top_acct_ls_ratio"]) + print(f" 조인 결과: {len(df)} rows (L/S 매칭: {len(valid)})") + print(f" top_acct_ls_ratio: mean={valid['top_acct_ls_ratio'].mean():.4f}, " + f"std={valid['top_acct_ls_ratio'].std():.4f}") + + # 백분위수 표시 + for p in [25, 50, 75]: + v = valid["top_acct_ls_ratio"].quantile(p / 100) + print(f" P{p}: {v:.4f}") + + # 2. 6개 조합 백테스트 + print("\n[2] 6개 조합 백테스트 실행") + print("-" * 80) + + combinations = [ + (75, "LONG", "모멘텀 강함: ratio ≥ P75 → LONG"), + (75, "SHORT", "역모멘텀: ratio ≥ P75 → SHORT"), + (50, "LONG", "모멘텀 중간: ratio ≥ P50 → LONG"), + (50, "SHORT", "역모멘텀 중간: ratio ≤ P50 → SHORT"), + (25, "LONG", "역모멘텀 약: ratio ≤ P25 → LONG"), + (25, "SHORT", "모멘텀 강함: ratio ≤ P25 → SHORT"), + ] + + results = [] + for pct, direction, desc in combinations: + # LONG: ratio >= threshold, SHORT: ratio <= threshold + # 25th percentile LONG = ratio가 낮을 때 LONG (Contrarian) + # 실제 로직: + # 75 LONG = ratio >= P75 (상위 25% 롱비율 높을 때 롱) = Momentum + # 75 SHORT = ratio >= P75 (상위 25% 롱비율 높을 때 숏) = Contrarian + # 25 SHORT = ratio <= P25 (하위 25% 롱비율 낮을 때 숏) = Momentum + # 25 LONG = ratio <= P25 (하위 25% 롱비율 낮을 때 롱) = Contrarian + + # 방향 보정: 25th에서 LONG은 "ratio <= P25일 때 LONG" (Contrarian) + if pct == 25 and direction == "LONG": + # 특수 케이스: 낮은 ratio에서 LONG (Contrarian) + result = run_backtest_contrarian(df, pct, "LONG") + elif pct == 25 and direction == "SHORT": + # ratio <= P25일 때 SHORT (Momentum) + result = run_backtest(df, pct, "SHORT") + elif pct == 75 and direction == "SHORT": + # ratio >= P75일 때 SHORT (Contrarian) + result = run_backtest_contrarian(df, pct, "SHORT") + else: + result = run_backtest(df, pct, direction) + + if result: + result["percentile"] = pct + result["direction"] = direction + result["description"] = desc + results.append(result) + else: + results.append({ + "percentile": pct, "direction": direction, + "description": desc, "trades": 0, "pf": 0, + "win_rate": 0, "total_pnl_bps": 0, "max_dd_bps": 0, + "threshold": 0, "wins": 0, "losses": 0, "avg_pnl_bps": 0, + }) + + # 3. 결과 테이블 + print("\n[3] 결과 테이블") + print("=" * 80) + print(f"{'조합':<35} {'거래수':>6} {'승률':>7} {'PF':>7} {'PnL(bps)':>10} {'MaxDD':>10} {'신뢰도':<15}") + print("-" * 80) + + for r in results: + emoji = confidence_emoji(r["trades"]) + label = confidence_label(r["trades"]) + pf_str = f"{r['pf']:.2f}" if r["pf"] != float("inf") else "INF" + print(f"{r['description']:<35} {r['trades']:>6} {r['win_rate']:>6.1f}% {pf_str:>7} " + f"{r['total_pnl_bps']:>+10.1f} {r['max_dd_bps']:>10.1f} {emoji} {label}") + + # 4. 필터 1: PF 판정 + print("\n[4] 필터 1: PF 판정") + print("-" * 80) + + strong = [r for r in results if r["pf"] > 1.5 and r["trades"] > 0] + weak = [r for r in results if 0.5 <= r["pf"] <= 1.5 and r["trades"] > 0] + failed = [r for r in results if r["pf"] < 0.5 and r["trades"] > 0] + + print(f" PF > 1.5 (명확한 edge): {len(strong)}개 조합") + for r in strong: + print(f" → {r['description']} (PF={r['pf']:.2f}, trades={r['trades']})") + print(f" 0.5 ≤ PF ≤ 1.5 (보류): {len(weak)}개 조합") + for r in weak: + print(f" ~ {r['description']} (PF={r['pf']:.2f}, trades={r['trades']})") + print(f" PF < 0.5 (실패): {len(failed)}개 조합") + for r in failed: + print(f" ✗ {r['description']} (PF={r['pf']:.2f}, trades={r['trades']})") + + # 5. 필터 2: 거래수 신뢰도 + print("\n[5] 필터 2: 거래수 신뢰도 (필터 1 통과 조합)") + print("-" * 80) + + filter2_passed = [r for r in strong if r["trades"] >= 20] + filter2_ref = [r for r in strong if r["trades"] < 20] + + if filter2_passed: + for r in filter2_passed: + print(f" ✓ {r['description']} — {r['trades']}건 ({confidence_label(r['trades'])})") + else: + print(" ⚠️ PF > 1.5 조합 중 거래수 20건 이상인 것 없음") + if filter2_ref: + for r in filter2_ref: + print(f" 🔴 {r['description']} — {r['trades']}건 (폐기: 과적합)") + + # 6. 필터 3: 대칭성 판정 + print("\n[6] 필터 3: 대칭성 판정") + print("-" * 80) + + # 같은 percentile에서 LONG/SHORT 양쪽 확인 + for pct in [75, 50, 25]: + long_r = next((r for r in results if r["percentile"] == pct and r["direction"] == "LONG"), None) + short_r = next((r for r in results if r["percentile"] == pct and r["direction"] == "SHORT"), None) + if not long_r or not short_r: + continue + + l_pf = long_r["pf"] + s_pf = short_r["pf"] + + if l_pf > 1.5 and s_pf > 1.5: + verdict = "Case 1: 양방향 생존 → ✓ Phase 2 후보" + elif (l_pf > 1.5 and s_pf < 0.5) or (s_pf > 1.5 and l_pf < 0.5): + verdict = "Case 2: 한쪽만 성공 → ✗ 시장 베타/우연 (폐기)" + elif l_pf > 1.5 or s_pf > 1.5: + verdict = "Case 3: 부분적 edge → ~ 낮은 신뢰도" + else: + verdict = "양쪽 모두 약함 → 해당 없음" + + print(f" P{pct}: LONG PF={l_pf:.2f}, SHORT PF={s_pf:.2f}") + print(f" → {verdict}") + + # 7. 최종 판정 + print("\n" + "=" * 80) + print(" [최종 판정]") + print("=" * 80) + + # Phase 2 후보 찾기 + phase2_candidates = [] + for pct in [75, 50, 25]: + long_r = next((r for r in results if r["percentile"] == pct and r["direction"] == "LONG"), None) + short_r = next((r for r in results if r["percentile"] == pct and r["direction"] == "SHORT"), None) + if not long_r or not short_r: + continue + + # Case 1: 양방향 PF > 1.5 + if long_r["pf"] > 1.5 and short_r["pf"] > 1.5: + if long_r["trades"] >= 20 and short_r["trades"] >= 20: + phase2_candidates.append(("Case1", pct, long_r, short_r)) + # Case 3: 한쪽만 PF > 1.5 + elif long_r["pf"] > 1.5 and long_r["trades"] >= 20: + phase2_candidates.append(("Case3-LONG", pct, long_r, short_r)) + elif short_r["pf"] > 1.5 and short_r["trades"] >= 20: + phase2_candidates.append(("Case3-SHORT", pct, long_r, short_r)) + + if phase2_candidates: + print("\n 🟢 Phase 2 진행 후보 발견!") + for case, pct, lr, sr in phase2_candidates: + print(f" [{case}] P{pct}: LONG PF={lr['pf']:.2f}({lr['trades']}건), " + f"SHORT PF={sr['pf']:.2f}({sr['trades']}건)") + print("\n → Phase 2 (Bot Simulation) 진행 권장") + print(" → 단, 8일 데이터이므로 4월 15일 재검증 필수") + else: + # 모든 조합 중 최고 PF + best = max(results, key=lambda r: r["pf"] if r["trades"] > 0 else 0) + if best["pf"] > 1.0: + print(f"\n 🟡 필터 미통과이나 PF > 1.0 조합 존재") + print(f" Best: {best['description']} (PF={best['pf']:.2f}, {best['trades']}건)") + print(f"\n → 데이터 부족. 4월 15일까지 수집 후 재검증") + else: + print(f"\n 🔴 PF > 1.0 조합 없음") + print(f" Best: {best['description']} (PF={best['pf']:.2f}, {best['trades']}건)") + print(f"\n → L/S ratio 단독 시그널로는 edge 없음") + print(f" → 다른 데이터 소스 탐색 권장") + + # 8. 추가: 전 구간 상세 (best 조합) + best = max(results, key=lambda r: r["pf"] if r["trades"] > 10 else 0) + if "df_trades" in best and best["trades"] > 0: + print(f"\n[참고] Best 조합 상세: {best['description']}") + print("-" * 60) + tdf = best["df_trades"] + print(f" 거래 기간: {tdf['entry_time'].min()} ~ {tdf['exit_time'].max()}") + print(f" 평균 진입 L/S ratio: {tdf['entry_ls_ratio'].mean():.4f}") + print(f" 수익 거래 평균: {tdf[tdf['net_return_bps']>0]['net_return_bps'].mean():.1f} bps") + if len(tdf[tdf['net_return_bps'] <= 0]) > 0: + print(f" 손실 거래 평균: {tdf[tdf['net_return_bps']<=0]['net_return_bps'].mean():.1f} bps") + print(f" 최대 연승: ", end="") + streaks = [] + streak = 0 + for _, row in tdf.iterrows(): + if row["net_return_bps"] > 0: + streak += 1 + else: + if streak > 0: + streaks.append(streak) + streak = 0 + if streak > 0: + streaks.append(streak) + print(f"{max(streaks) if streaks else 0}연승") + + print("\n" + "=" * 80) + print(" 분석 완료. 결과를 바탕으로 의사결정하세요.") + print("=" * 80) + + +def run_backtest_contrarian(df, percentile, direction, hold_bars=HOLD_BARS): + """ + Contrarian 방향 백테스트. + - P25 + LONG: ratio <= P25일 때 LONG (낮은 ratio에서 롱) + - P75 + SHORT: ratio >= P75일 때 SHORT (높은 ratio에서 숏) + """ + threshold = df["top_acct_ls_ratio"].quantile(percentile / 100) + + trades = [] + i = 0 + while i < len(df) - hold_bars: + ratio = df.iloc[i]["top_acct_ls_ratio"] + if pd.isna(ratio): + i += 1 + continue + + trigger = False + if direction == "LONG" and ratio <= threshold: + trigger = True + elif direction == "SHORT" and ratio >= threshold: + trigger = True + + if trigger: + entry_price = df.iloc[i + 1]["open"] + exit_price = df.iloc[i + 1 + hold_bars - 1]["close"] + if direction == "LONG": + gross_return = (exit_price / entry_price) - 1 + else: + gross_return = (entry_price / exit_price) - 1 + fee = FEE_RATE * 2 + net_return = gross_return - fee + trades.append({ + "entry_time": df.iloc[i + 1]["timestamp"], + "exit_time": df.iloc[i + 1 + hold_bars - 1]["timestamp"], + "entry_price": entry_price, + "exit_price": exit_price, + "entry_ls_ratio": ratio, + "gross_return_bps": gross_return * 10000, + "net_return_bps": net_return * 10000, + "fee_bps": fee * 10000, + }) + i += 1 + hold_bars + else: + i += 1 + + if not trades: + return None + + df_trades = pd.DataFrame(trades) + wins = df_trades[df_trades["net_return_bps"] > 0]["net_return_bps"] + losses = df_trades[df_trades["net_return_bps"] <= 0]["net_return_bps"] + + gross_profit = wins.sum() if len(wins) > 0 else 0 + gross_loss = abs(losses.sum()) if len(losses) > 0 else 0 + pf = gross_profit / gross_loss if gross_loss > 0 else float("inf") if gross_profit > 0 else 0 + + cum_pnl = df_trades["net_return_bps"].cumsum() + running_max = cum_pnl.cummax() + max_dd = (cum_pnl - running_max).min() + + return { + "trades": len(df_trades), + "wins": len(wins), + "losses": len(losses), + "win_rate": len(wins) / len(df_trades) * 100, + "pf": pf, + "total_pnl_bps": df_trades["net_return_bps"].sum(), + "avg_pnl_bps": df_trades["net_return_bps"].mean(), + "max_dd_bps": max_dd, + "threshold": threshold, + "df_trades": df_trades, + } + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/scripts/revalidate_apr15.py b/scripts/revalidate_apr15.py new file mode 100644 index 0000000..2a231f8 --- /dev/null +++ b/scripts/revalidate_apr15.py @@ -0,0 +1,49 @@ +""" +4월 15일 재검증 스크립트 — L/S ratio + FR×OI 동시 재실행 + +crontab: 0 10 15 4 * cd /root/cointrader && /root/cointrader/.venv/bin/python scripts/revalidate_apr15.py + +재검증 대상: +1. L/S ratio (top_acct_ls_ratio) — 24일 데이터로 6개 조합 +2. FR × OI변화율(1h) — 29일 데이터로 12개 조합 +3. 대칭성 재판정 + +Usage: python scripts/revalidate_apr15.py +""" + +import subprocess +import sys +from datetime import datetime, timezone + +def main(): + now = datetime.now(timezone.utc) + print("=" * 80) + print(f" 4월 재검증 실행 — {now.strftime('%Y-%m-%d %H:%M UTC')}") + print("=" * 80) + + print("\n[1/2] L/S ratio 백테스트 재실행") + print("-" * 40) + r1 = subprocess.run( + [sys.executable, "scripts/ls_ratio_backtest.py"], + capture_output=False, + ) + + print("\n\n[2/2] FR × OI 백테스트 재실행") + print("-" * 40) + r2 = subprocess.run( + [sys.executable, "scripts/fr_oi_backtest.py"], + capture_output=False, + ) + + print("\n" + "=" * 80) + print(" 재검증 완료") + print("=" * 80) + print(f"\n L/S ratio: {'성공' if r1.returncode == 0 else '실패'}") + print(f" FR × OI: {'성공' if r2.returncode == 0 else '실패'}") + print(f"\n 판정 기준:") + print(f" - L/S ratio: PF > 1.0인 조합 있으면 재검토") + print(f" - FR × OI: SHORT+LONG 모두 PF > 1.0이면 대칭성 통과") + print(f" - 둘 다 실패 시 확정 폐기") + +if __name__ == "__main__": + main() diff --git a/scripts/trade_ls_analysis.py b/scripts/trade_ls_analysis.py new file mode 100644 index 0000000..954f0fa --- /dev/null +++ b/scripts/trade_ls_analysis.py @@ -0,0 +1,459 @@ +""" +Trade History + L/S Ratio 종합 분석 +- 봇 대시보드 API에서 거래 기록 로드 +- Binance API에서 L/S ratio (30일) 로드 + 로컬 parquet 병합 +- 진입/청산 시점 L/S ratio 매칭 +- 수익/손실 거래별 L/S 분포 분석 +- L/S 임계값 필터링 시뮬레이션 + +Usage: python scripts/trade_ls_analysis.py [--api URL] +""" + +import asyncio +import aiohttp +import pandas as pd +import numpy as np +from datetime import datetime, timedelta, timezone +from pathlib import Path +import argparse +import json + +BASE = "https://fapi.binance.com" +DASHBOARD_API = "http://10.1.10.24:8080/api/trades" +DATA_DIR = Path("data") +SYMBOLS_FOR_LS = ["XRPUSDT", "BTCUSDT", "ETHUSDT"] + + +async def fetch_json(session, url, params=None): + async with session.get(url, params=params) as resp: + return await resp.json() + + +async def fetch_ls_ratios_from_api(session, symbol, start_ms, end_ms): + """Binance API에서 L/S ratio 전체 기간 가져오기 (페이징)""" + all_top_acct = [] + all_global = [] + + for endpoint, target in [ + (f"{BASE}/futures/data/topLongShortAccountRatio", all_top_acct), + (f"{BASE}/futures/data/globalLongShortAccountRatio", all_global), + ]: + current = start_ms + while current < end_ms: + params = { + "symbol": symbol, + "period": "15m", + "startTime": current, + "endTime": end_ms, + "limit": 500, + } + data = await fetch_json(session, endpoint, params) + if not data or not isinstance(data, list): + break + target.extend(data) + last_ts = int(data[-1]["timestamp"]) + if last_ts <= current: + break + current = last_ts + 1 + + def to_df(data, col_name): + if not data: + return pd.DataFrame() + df = pd.DataFrame(data) + df["timestamp"] = pd.to_datetime(df["timestamp"].astype(int), unit="ms", utc=True) + df[col_name] = df["longShortRatio"].astype(float) + return df[["timestamp", col_name]].drop_duplicates("timestamp") + + df_top = to_df(all_top_acct, "top_acct_ls_ratio") + df_global = to_df(all_global, "global_ls_ratio") + + if df_top.empty and df_global.empty: + return pd.DataFrame() + + if df_top.empty: + return df_global + if df_global.empty: + return df_top + + return df_top.merge(df_global, on="timestamp", how="outer").sort_values("timestamp") + + +def load_local_ls_ratio(symbol): + """로컬 parquet에서 L/S ratio 로드""" + path = DATA_DIR / symbol.lower() / "ls_ratio_15m.parquet" + if not path.exists(): + return pd.DataFrame() + df = pd.read_parquet(path) + if "timestamp" in df.columns: + df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True) + return df + + +def find_nearest_ls(ls_df, target_time, max_gap_minutes=30): + """타겟 시간에 가장 가까운 L/S ratio 찾기""" + if ls_df.empty: + return None, None, None + + target = pd.Timestamp(target_time, tz="UTC") + diffs = (ls_df["timestamp"] - target).abs() + idx = diffs.idxmin() + gap = diffs[idx] + + if gap > pd.Timedelta(minutes=max_gap_minutes): + return None, None, gap + + row = ls_df.loc[idx] + return row.get("top_acct_ls_ratio"), row.get("global_ls_ratio"), gap + + +def classify_signal(trade): + """진입 신호 분류""" + rsi = trade.get("rsi", 0) + macd = trade.get("macd_hist", 0) + direction = trade["direction"] + + signals = [] + if direction == "LONG": + if rsi and rsi > 65: + signals.append("RSI과매수진입") + elif rsi and rsi < 35: + signals.append("RSI역방향") + if macd and macd > 0: + signals.append("MACD+") + elif macd and macd < 0: + signals.append("MACD역방향") + else: # SHORT + if rsi and rsi < 35: + signals.append("RSI과매도진입") + elif rsi and rsi > 65: + signals.append("RSI역방향") + if macd and macd < 0: + signals.append("MACD-") + elif macd and macd > 0: + signals.append("MACD역방향") + + return ", ".join(signals) if signals else "복합신호" + + +def classify_close_reason(trade): + """청산 이유 분류""" + reason = trade["close_reason"] + if reason == "TP": + return "TP(익절)" + elif reason == "SYNC": + return "SL(손절)" + elif reason == "MANUAL": + # MANUAL인데 SL가격과 exit가격이 같으면 SL + sl = trade.get("sl") + exit_p = trade.get("exit_price") + if sl and exit_p and abs(float(sl) - float(exit_p)) < 0.0001: + return "SL(손절)" + # 역방향 시그널로 청산 + extra = trade.get("extra", "{}") + if isinstance(extra, str): + try: + extra = json.loads(extra) + except json.JSONDecodeError: + extra = {} + if extra.get("recovery"): + return "신호반전" + return "SL(손절)" # 대부분 MANUAL은 SL 히트 + return reason + + +async def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--api", default=DASHBOARD_API) + args = parser.parse_args() + + print("=" * 80) + print(" Trade History + L/S Ratio 종합 분석") + print("=" * 80) + + # 1. 거래 데이터 로드 + async with aiohttp.ClientSession() as session: + trade_data = await fetch_json(session, args.api) + trades = trade_data["trades"] + print(f"\n📊 거래 데이터: {len(trades)}건 로드") + + # 2. L/S ratio 데이터 로드 (API + local) + # 가장 오래된 거래 기준으로 시작 시간 설정 + earliest = min(t["entry_time"] for t in trades) + start_dt = pd.Timestamp(earliest, tz="UTC") - timedelta(hours=1) + end_dt = datetime.now(timezone.utc) + start_ms = int(start_dt.timestamp() * 1000) + end_ms = int(end_dt.timestamp() * 1000) + + print(f"📡 Binance API에서 L/S ratio 로딩 ({start_dt.date()} ~ {end_dt.date()})...") + + ls_data = {} + for sym in SYMBOLS_FOR_LS: + api_df = await fetch_ls_ratios_from_api(session, sym, start_ms, end_ms) + local_df = load_local_ls_ratio(sym) + + if not api_df.empty and not local_df.empty: + combined = pd.concat([api_df, local_df]).drop_duplicates("timestamp").sort_values("timestamp") + elif not api_df.empty: + combined = api_df + elif not local_df.empty: + combined = local_df + else: + combined = pd.DataFrame() + + ls_data[sym] = combined.reset_index(drop=True) + print(f" {sym}: {len(ls_data[sym])} rows ({ls_data[sym]['timestamp'].min()} ~ {ls_data[sym]['timestamp'].max()})" if not combined.empty else f" {sym}: no data") + + # 3. 거래별 L/S ratio 매칭 + print("\n" + "=" * 80) + print(" 1. 거래 기록 + L/S Ratio 매칭") + print("=" * 80) + + enriched = [] + for t in trades: + sym = t["symbol"] + # XRP 거래에는 XRP L/S, 다른 심볼도 XRP L/S 참조 (크로스 분석) + ls_sym = ls_data.get(sym, pd.DataFrame()) + ls_xrp = ls_data.get("XRPUSDT", pd.DataFrame()) + ls_btc = ls_data.get("BTCUSDT", pd.DataFrame()) + + entry_top, entry_global, _ = find_nearest_ls(ls_sym if not ls_sym.empty else ls_xrp, t["entry_time"]) + exit_top, exit_global, _ = find_nearest_ls(ls_sym if not ls_sym.empty else ls_xrp, t["exit_time"]) + + # BTC L/S for cross-reference + btc_entry_top, btc_entry_global, _ = find_nearest_ls(ls_btc, t["entry_time"]) + + enriched.append({ + "id": t["id"], + "symbol": sym, + "direction": t["direction"], + "entry_time": t["entry_time"], + "exit_time": t["exit_time"], + "signal": classify_signal(t), + "close_reason": classify_close_reason(t), + "rsi": t.get("rsi"), + "macd_hist": t.get("macd_hist"), + "entry_top_acct_ls": entry_top, + "entry_global_ls": entry_global, + "exit_top_acct_ls": exit_top, + "exit_global_ls": exit_global, + "ls_change_top": (exit_top - entry_top) if entry_top and exit_top else None, + "ls_change_global": (exit_global - entry_global) if entry_global and exit_global else None, + "btc_entry_top_ls": btc_entry_top, + "btc_entry_global_ls": btc_entry_global, + "net_pnl": t["net_pnl"], + "is_win": t["net_pnl"] > 0, + "entry_price": t["entry_price"], + "exit_price": t["exit_price"], + }) + + df = pd.DataFrame(enriched) + + # 거래 기록 테이블 출력 + print(f"\n{'ID':>3} {'심볼':<10} {'방향':<5} {'진입시간':<20} {'진입신호':<16} " + f"{'진입L/S':>7} {'청산L/S':>7} {'ΔL/S':>7} {'청산이유':<10} {'PnL':>8}") + print("-" * 120) + for _, r in df.iterrows(): + entry_ls = f"{r['entry_top_acct_ls']:.3f}" if pd.notna(r['entry_top_acct_ls']) else "N/A" + exit_ls = f"{r['exit_top_acct_ls']:.3f}" if pd.notna(r['exit_top_acct_ls']) else "N/A" + delta_ls = f"{r['ls_change_top']:+.3f}" if pd.notna(r['ls_change_top']) else "N/A" + pnl_str = f"{r['net_pnl']:+.4f}" + print(f"{r['id']:>3} {r['symbol']:<10} {r['direction']:<5} {r['entry_time']:<20} {r['signal']:<16} " + f"{entry_ls:>7} {exit_ls:>7} {delta_ls:>7} {r['close_reason']:<10} {pnl_str:>8}") + + # 4. 수익 거래 vs 손실 거래 L/S 비교 + print("\n" + "=" * 80) + print(" 2. 수익 거래 vs 손실 거래: L/S Ratio 비교") + print("=" * 80) + + has_ls = df.dropna(subset=["entry_top_acct_ls"]) + if len(has_ls) > 0: + wins = has_ls[has_ls["is_win"]] + losses = has_ls[~has_ls["is_win"]] + + print(f"\n L/S ratio 매칭된 거래: {len(has_ls)}건 (수익: {len(wins)}, 손실: {len(losses)})") + print(f"\n {'지표':<30} {'수익 거래':>12} {'손실 거래':>12} {'차이':>10}") + print(" " + "-" * 70) + + for col, label in [ + ("entry_top_acct_ls", "진입 시 top_acct L/S"), + ("entry_global_ls", "진입 시 global L/S"), + ("exit_top_acct_ls", "청산 시 top_acct L/S"), + ("exit_global_ls", "청산 시 global L/S"), + ("ls_change_top", "진입→청산 ΔL/S (top)"), + ("ls_change_global", "진입→청산 ΔL/S (global)"), + ("btc_entry_top_ls", "BTC 진입 시 top_acct L/S"), + ]: + w_vals = wins[col].dropna() + l_vals = losses[col].dropna() + if len(w_vals) > 0 and len(l_vals) > 0: + w_mean = w_vals.mean() + l_mean = l_vals.mean() + diff = w_mean - l_mean + print(f" {label:<30} {w_mean:>12.4f} {l_mean:>12.4f} {diff:>+10.4f}") + else: + w_str = f"{w_vals.mean():.4f}" if len(w_vals) > 0 else "N/A" + l_str = f"{l_vals.mean():.4f}" if len(l_vals) > 0 else "N/A" + print(f" {label:<30} {w_str:>12} {l_str:>12} {'N/A':>10}") + else: + print("\n ⚠️ L/S ratio 매칭 가능한 거래가 없습니다") + + # 5. 진입 시점 L/S와 거래 결과의 상관계수 + print("\n" + "=" * 80) + print(" 3. L/S Ratio ↔ PnL 상관계수") + print("=" * 80) + + for col, label in [ + ("entry_top_acct_ls", "진입 top_acct L/S"), + ("entry_global_ls", "진입 global L/S"), + ("btc_entry_top_ls", "BTC 진입 top_acct L/S"), + ("ls_change_top", "ΔL/S (top)"), + ]: + valid = df.dropna(subset=[col, "net_pnl"]) + if len(valid) >= 3: + corr = valid[col].corr(valid["net_pnl"]) + print(f" {label:<30} r = {corr:>+.4f} (n={len(valid)})") + else: + print(f" {label:<30} 데이터 부족 (n={len(valid)})") + + # 6. 방향별 분석 (LONG 진입 시 L/S 높으면? SHORT 진입 시 낮으면?) + print("\n" + "=" * 80) + print(" 4. 방향별 L/S Ratio 분석") + print("=" * 80) + + for direction in ["LONG", "SHORT"]: + subset = has_ls[has_ls["direction"] == direction] + if len(subset) == 0: + continue + print(f"\n [{direction}] ({len(subset)}건)") + wins_d = subset[subset["is_win"]] + losses_d = subset[~subset["is_win"]] + print(f" 수익: {len(wins_d)}건, 손실: {len(losses_d)}건") + if len(subset) > 0: + for col in ["entry_top_acct_ls", "entry_global_ls"]: + vals = subset[col].dropna() + if len(vals) > 0: + w = wins_d[col].dropna() + l = losses_d[col].dropna() + w_str = f"{w.mean():.4f}" if len(w) > 0 else "N/A" + l_str = f"{l.mean():.4f}" if len(l) > 0 else "N/A" + print(f" {col}: 수익평균={w_str}, 손실평균={l_str}") + + # 7. 청산 이유별 L/S ratio 분포 + print("\n" + "=" * 80) + print(" 5. 청산 이유별 L/S Ratio 분포") + print("=" * 80) + + for reason in df["close_reason"].unique(): + subset = has_ls[has_ls["close_reason"] == reason] + if len(subset) == 0: + continue + print(f"\n [{reason}] ({len(subset)}건)") + for col, label in [("entry_top_acct_ls", "진입 L/S"), ("exit_top_acct_ls", "청산 L/S"), ("ls_change_top", "ΔL/S")]: + vals = subset[col].dropna() + if len(vals) > 0: + print(f" {label}: mean={vals.mean():.4f}, std={vals.std():.4f}, min={vals.min():.4f}, max={vals.max():.4f}") + + # 8. L/S 임계값 필터링 시뮬레이션 + print("\n" + "=" * 80) + print(" 6. L/S 임계값 필터링 시뮬레이션") + print(" '만약 L/S 조건으로 진입을 필터링했다면?'") + print("=" * 80) + + if len(has_ls) > 0: + # 시뮬레이션 1: top_acct L/S ratio 기준 필터 + print("\n [A] top_acct_ls_ratio 임계값별 (LONG 진입 시 ratio > threshold)") + print(f" {'Threshold':>10} {'통과':>5} {'차단':>5} {'통과 PnL':>10} {'차단 PnL':>10} {'통과 승률':>10} {'원본 승률':>10}") + print(" " + "-" * 70) + + longs = has_ls[has_ls["direction"] == "LONG"] + shorts = has_ls[has_ls["direction"] == "SHORT"] + all_wr = has_ls["is_win"].mean() * 100 if len(has_ls) > 0 else 0 + + if len(longs) > 0: + ls_vals = longs["entry_top_acct_ls"].dropna() + if len(ls_vals) > 0: + for pct in [0.25, 0.50, 0.75]: + threshold = ls_vals.quantile(pct) + passed = longs[longs["entry_top_acct_ls"] >= threshold] + blocked = longs[longs["entry_top_acct_ls"] < threshold] + p_pnl = passed["net_pnl"].sum() + b_pnl = blocked["net_pnl"].sum() + p_wr = passed["is_win"].mean() * 100 if len(passed) > 0 else 0 + print(f" {threshold:>10.4f} {len(passed):>5} {len(blocked):>5} " + f"{p_pnl:>+10.4f} {b_pnl:>+10.4f} {p_wr:>9.1f}% {all_wr:>9.1f}%") + + # 시뮬레이션 2: SHORT 진입 시 ratio < threshold + print(f"\n [B] top_acct_ls_ratio 임계값별 (SHORT 진입 시 ratio < threshold)") + print(f" {'Threshold':>10} {'통과':>5} {'차단':>5} {'통과 PnL':>10} {'차단 PnL':>10} {'통과 승률':>10}") + print(" " + "-" * 70) + + if len(shorts) > 0: + ls_vals = shorts["entry_top_acct_ls"].dropna() + if len(ls_vals) > 0: + for pct in [0.75, 0.50, 0.25]: + threshold = ls_vals.quantile(pct) + passed = shorts[shorts["entry_top_acct_ls"] <= threshold] + blocked = shorts[shorts["entry_top_acct_ls"] > threshold] + p_pnl = passed["net_pnl"].sum() + b_pnl = blocked["net_pnl"].sum() + p_wr = passed["is_win"].mean() * 100 if len(passed) > 0 else 0 + print(f" {threshold:>10.4f} {len(passed):>5} {len(blocked):>5} " + f"{p_pnl:>+10.4f} {b_pnl:>+10.4f} {p_wr:>9.1f}%") + + # 시뮬레이션 3: Momentum 전략 - L/S 방향과 같은 방향만 진입 + print(f"\n [C] Momentum 필터: L/S ratio > 중앙값이면 LONG만, < 중앙값이면 SHORT만") + if len(has_ls) > 0: + median_ls = has_ls["entry_top_acct_ls"].median() + momentum_filter = has_ls.apply( + lambda r: (r["direction"] == "LONG" and r["entry_top_acct_ls"] >= median_ls) or + (r["direction"] == "SHORT" and r["entry_top_acct_ls"] < median_ls), + axis=1 + ) + passed = has_ls[momentum_filter] + blocked = has_ls[~momentum_filter] + print(f" 중앙값: {median_ls:.4f}") + print(f" 통과: {len(passed)}건, PnL합계: {passed['net_pnl'].sum():+.4f}, " + f"승률: {passed['is_win'].mean()*100:.1f}%") + print(f" 차단: {len(blocked)}건, PnL합계: {blocked['net_pnl'].sum():+.4f}, " + f"승률: {blocked['is_win'].mean()*100:.1f}%") + + # 시뮬레이션 4: Contrarian 전략 - L/S 반대 방향만 진입 + print(f"\n [D] Contrarian 필터: L/S ratio > 중앙값이면 SHORT만, < 중앙값이면 LONG만") + if len(has_ls) > 0: + contrarian_filter = has_ls.apply( + lambda r: (r["direction"] == "SHORT" and r["entry_top_acct_ls"] >= median_ls) or + (r["direction"] == "LONG" and r["entry_top_acct_ls"] < median_ls), + axis=1 + ) + passed = has_ls[contrarian_filter] + blocked = has_ls[~contrarian_filter] + print(f" 통과: {len(passed)}건, PnL합계: {passed['net_pnl'].sum():+.4f}, " + f"승률: {passed['is_win'].mean()*100:.1f}%") + print(f" 차단: {len(blocked)}건, PnL합계: {blocked['net_pnl'].sum():+.4f}, " + f"승률: {blocked['is_win'].mean()*100:.1f}%") + + # 9. 전체 L/S ratio 시계열 + 거래 오버레이 요약 + print("\n" + "=" * 80) + print(" 7. 전체 요약") + print("=" * 80) + + total_trades = len(df) + ls_matched = len(has_ls) + total_pnl = df["net_pnl"].sum() + win_rate = df["is_win"].mean() * 100 + print(f"\n 전체 거래: {total_trades}건 (L/S 매칭: {ls_matched}건)") + print(f" 총 PnL: {total_pnl:+.4f} USDT") + print(f" 승률: {win_rate:.1f}% ({df['is_win'].sum()}/{total_trades})") + + if len(has_ls) > 0: + print(f"\n L/S 매칭 거래 통계:") + print(f" 진입 top_acct L/S 범위: {has_ls['entry_top_acct_ls'].min():.4f} ~ {has_ls['entry_top_acct_ls'].max():.4f}") + print(f" 진입 global L/S 범위: {has_ls['entry_global_ls'].min():.4f} ~ {has_ls['entry_global_ls'].max():.4f}") + + print(f"\n ⚠️ 주의: 거래 {total_trades}건은 통계적 유의성이 부족합니다.") + print(f" 현재 결과는 탐색적 분석이며, 최소 50건 이상의 거래가 필요합니다.") + print(f" L/S ratio 데이터는 계속 축적 중이므로 4월 말 재분석을 권장합니다.") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_evaluate_oos.py b/tests/test_evaluate_oos.py new file mode 100644 index 0000000..fbb10f9 --- /dev/null +++ b/tests/test_evaluate_oos.py @@ -0,0 +1,181 @@ +""" +evaluate_oos.py 비용 모델 단위 테스트 +""" + +import sys +from pathlib import Path + +import pandas as pd +import pytest + +# 프로젝트 루트를 path에 추가 +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from scripts.evaluate_oos import ( + apply_cost_model, + calc_metrics, + calc_trade_cost, + count_funding_events, +) +from src.config import COST_MODEL, COST_SCENARIOS + + +# ── count_funding_events 테스트 ────────────────────────────────── + + +def test_count_funding_events_no_crossing(): + """진입 01:00 UTC -> 청산 05:00 UTC, 펀딩 경계(00/08/16) 미포함 -> count == 0.""" + entry = pd.Timestamp("2026-04-10 01:00:00+00:00") + exit_ = pd.Timestamp("2026-04-10 05:00:00+00:00") + assert count_funding_events(entry, exit_) == 0 + + +def test_count_funding_events_single_crossing(): + """진입 06:00 UTC -> 청산 10:00 UTC, 08:00 포함 -> count == 1.""" + entry = pd.Timestamp("2026-04-10 06:00:00+00:00") + exit_ = pd.Timestamp("2026-04-10 10:00:00+00:00") + assert count_funding_events(entry, exit_) == 1 + + +def test_count_funding_events_multiple_crossings(): + """12시간 보유: 02:00 -> 14:00, 08:00 포함 -> count == 1.""" + entry = pd.Timestamp("2026-04-10 02:00:00+00:00") + exit_ = pd.Timestamp("2026-04-10 14:00:00+00:00") + assert count_funding_events(entry, exit_) == 1 + + # 22:00 -> 10:00 (다음날), 00:00 + 08:00 포함 -> count == 2 + entry2 = pd.Timestamp("2026-04-10 22:00:00+00:00") + exit2 = pd.Timestamp("2026-04-11 10:00:00+00:00") + assert count_funding_events(entry2, exit2) == 2 + + +def test_count_funding_events_short_trade_no_overcounting(): + """75분 거래, 경계 미포함 -> count == 0.""" + # 18:15 -> 19:30, 펀딩 경계 없음 + entry = pd.Timestamp("2026-04-10 18:15:00+00:00") + exit_ = pd.Timestamp("2026-04-10 19:30:00+00:00") + assert count_funding_events(entry, exit_) == 0 + + +def test_count_funding_events_exact_boundary(): + """정확히 경계에서 진입/청산하는 경우.""" + # entry=08:00, exit=16:00 -> ceil(08:00)=08:00, floor(16:00)=16:00 + # hours: 08, 09, ..., 16 -> 08:00(yes), 16:00(yes) -> count == 2 + entry = pd.Timestamp("2026-04-10 08:00:00+00:00") + exit_ = pd.Timestamp("2026-04-10 16:00:00+00:00") + assert count_funding_events(entry, exit_) == 2 + + +# ── 비용 계산 테스트 ───────────────────────────────────────────── + + +def test_cost_calculation_taker_roundtrip(): + """진입 taker + SL taker, slippage 0, funding 0 -> 8 bps.""" + row = pd.Series({ + "entry_ts": pd.Timestamp("2026-04-10 01:00:00+00:00"), + "exit_ts": pd.Timestamp("2026-04-10 02:00:00+00:00"), + "pnl_bps": -50.0, + "reason": "SL 히트 (1.3012)", + "side": "SHORT", + }) + scenario = COST_SCENARIOS["fees_only"] + cost = calc_trade_cost(row, scenario) + assert cost == 8.0 # taker(4) + taker(4) + 0 + 0 + + +def test_cost_calculation_tp_exit(): + """TP 히트 시에도 현재 설정에서는 taker -> 8 bps.""" + row = pd.Series({ + "entry_ts": pd.Timestamp("2026-04-10 01:00:00+00:00"), + "exit_ts": pd.Timestamp("2026-04-10 02:00:00+00:00"), + "pnl_bps": 80.0, + "reason": "TP 히트 (1.3826)", + "side": "LONG", + }) + scenario = COST_SCENARIOS["fees_only"] + cost = calc_trade_cost(row, scenario) + assert cost == 8.0 + + +def test_cost_with_slippage_and_funding(): + """realistic 시나리오: fee 8 + slippage 2 + funding 1 = 11 bps.""" + # 진입 15:45, 청산 17:00 -> funding event at 16:00 -> count=1 + row = pd.Series({ + "entry_ts": pd.Timestamp("2026-04-02 15:45:00+00:00"), + "exit_ts": pd.Timestamp("2026-04-02 17:00:00+00:00"), + "pnl_bps": -68.0, + "reason": "SL 히트 (1.3012)", + "side": "SHORT", + }) + scenario = COST_SCENARIOS["realistic"] + cost = calc_trade_cost(row, scenario) + # fee=8, slippage=1*2=2, funding=1*1=1 -> total=11 + assert cost == 11.0 + + +def test_adjusted_pnl_matches_manual(): + """첫 번째 거래(Trade #0)에 대해 수작업 계산값과 일치 확인.""" + # Trade #0: SHORT, entry 15:45 UTC, exit 17:00 UTC, pnl_bps=-68.0, SL 히트 + # fees_only: cost=8 (fee only, funding event at 16:00 but funding_bps=0) -> adjusted=-76.0 + # realistic: cost=8+2+1=11 -> adjusted=-79.0 + # pessimistic: cost=8+6+2=16 -> adjusted=-84.0 + row = pd.Series({ + "entry_ts": pd.Timestamp("2026-04-02 15:45:02.285284+00:00"), + "exit_ts": pd.Timestamp("2026-04-02 17:00:00.791551+00:00"), + "pnl_bps": -68.0, + "reason": "SL 히트 (1.3012)", + "side": "SHORT", + }) + + for scenario_name, expected_adj in [ + ("fees_only", -76.0), + ("realistic", -79.0), + ("pessimistic", -84.0), + ]: + scenario = COST_SCENARIOS[scenario_name] + cost = calc_trade_cost(row, scenario) + adjusted = row["pnl_bps"] - cost + assert adjusted == expected_adj, f"{scenario_name}: {adjusted} != {expected_adj}" + + +# ── 회귀 테스트 ────────────────────────────────────────────────── + + +def test_regression_fees_only_cum_pnl(): + """18건 전체를 fees_only로 돌렸을 때 CumPnL == -173.9 bps (+-0.5 bps 허용).""" + jsonl_path = Path("data/trade_history/mtf_xrpusdtusdt.jsonl") + if not jsonl_path.exists(): + pytest.skip("로컬 jsonl 파일 없음") + + df = pd.read_json(jsonl_path, lines=True) + df["entry_ts"] = pd.to_datetime(df["entry_ts"], utc=True) + df["exit_ts"] = pd.to_datetime(df["exit_ts"], utc=True) + df["duration_min"] = (df["exit_ts"] - df["entry_ts"]).dt.total_seconds() / 60 + + result = apply_cost_model(df, "fees_only") + metrics = calc_metrics(result, pnl_col="adjusted_pnl_bps") + + assert metrics["trades"] == 18 + assert abs(metrics["cum_pnl"] - (-173.9)) <= 0.5, f"CumPnL={metrics['cum_pnl']}, expected -173.9" + + +# ── calc_metrics 테스트 ────────────────────────────────────────── + + +def test_calc_metrics_empty(): + """빈 DataFrame -> 안전한 기본값.""" + df = pd.DataFrame(columns=["pnl_bps", "duration_min"]) + m = calc_metrics(df) + assert m["trades"] == 0 + assert m["pf"] == 0.0 + + +def test_calc_metrics_with_avg_pnl(): + """avg_pnl 필드가 정확히 계산되는지 확인.""" + df = pd.DataFrame({ + "pnl_bps": [10.0, -5.0, 20.0], + "duration_min": [60.0, 30.0, 90.0], + }) + m = calc_metrics(df) + assert m["trades"] == 3 + assert m["avg_pnl"] == pytest.approx(25.0 / 3, abs=0.01)