diff --git a/.env.example b/.env.example index e8c52e7..6d0c7d0 100644 --- a/.env.example +++ b/.env.example @@ -77,16 +77,22 @@ CAUSAL_SIM_REPORT_HTML=docs/spot/2_analysis/causal_sim_report.html # --- 현물 3단계: 운영 (기본 paper) --- OPS_MODE=paper -OPS_TECHNIQUE_ID=composite_v3 -# OPS_MIN_SCORE=3.5 -OPS_MTF_ENABLED=true -OPS_TREND_GATE_ENABLED=true -OPS_DAILY_MAX_TRADES=20 +OPS_TECHNIQUE_ID=fractal_swing +OPS_MTF_ENABLED=false +OPS_TREND_GATE_ENABLED=false +OPS_DAILY_MAX_TRADES=100 OPS_MIN_ORDER_KRW=5000 +OPS_SLIPPAGE_RATE=0.0005 +OPS_ORDER_INTERVAL_SEC=0.35 OPS_SYNC_CANDLES=true -OPS_STATE_JSON=data/spot/operations/ops_state.json -OPS_REPORT_JSON=docs/spot/3_operations/ops_report.json -OPS_FILTERED_BACKTEST_JSON=docs/spot/3_operations/filtered_backtest_report.json +# 비우면 DOWNLOAD_INTERVALS 전체 증분 sync +# OPS_SYNC_INTERVALS= +OPS_SIGNAL_TAIL_BARS=800 +# OPS_PERSIST_SIGNAL_CACHE=false +OPS_STATE_JSON=data/spot/operations/fractal_ops_state.json +OPS_REPORT_JSON=docs/spot/3_operations/fractal_ops_report.json +OPS_FILTERED_BACKTEST_JSON=docs/spot/3_operations/fractal_filtered_backtest_report.json +# composite_v3 운영 시: OPS_TECHNIQUE_ID=composite_v3, OPS_MIN_SCORE=2.5, OPS_MTF_ENABLED=true # 폴더 구조: data|docs / {common, spot, futures} # common — coins.db 등 공유 리소스 diff --git a/.gitignore b/.gitignore index e8578e4..7b8fc8b 100644 --- a/.gitignore +++ b/.gitignore @@ -20,6 +20,9 @@ !/docs/spot/3_operations/ /docs/spot/3_operations/** !docs/spot/3_operations/stage3_design_guide.md +!/docs/live/ +/docs/live/** +!docs/live/index.html logs/ *.db diff --git a/README.md b/README.md index 8fad230..49260fe 100644 --- a/README.md +++ b/README.md @@ -51,19 +51,21 @@ DeepCoin/ │ ├── spot/ # 현물 전용 데이터 │ │ ├── ground_truth/ # 0단계 GT JSON │ │ ├── techniques/ # 2단계 기법 결과 -│ │ └── mtf/ # 2단계 MTF 규칙 +│ │ ├── mtf/ # 2단계 MTF 규칙 +│ │ └── operations/ # 3단계 운영 상태 │ └── futures/ # 선물 전용 데이터 │ ├── ground_truth/ # 0단계 선물 GT JSON │ ├── techniques/ # (예정) 2단계 │ └── mtf/ # (예정) 2단계 │ └── docs/ + ├── live/ # live 운영 백테스트 매매 차트 ├── common/ # 공통 문서 (예정) ├── spot/ # 현물 리포트·차트 │ ├── 0_ground_truth/ # 0단계 GT 차트 │ ├── 1_simulation/ # 1단계 sim 차트 │ ├── 2_analysis/ # 2단계 분석 리포트 - │ └── 3_operations/ # 3단계 운영 (예정) + │ └── 3_operations/ # 3단계 운영 리포트·백테스트 └── futures/ # 선물 리포트·차트 ├── 0_ground_truth/ # 0단계 선물 GT 차트 ├── 1_simulation/ # (예정) 1단계 @@ -125,10 +127,11 @@ bash scripts/2_run_stage2_all.sh # ── futures 0단계: 선물 GT (현물 GT 기반) ─────────────────── python scripts/0_ground_truth_futures.py --tier all -# ── spot 3단계: 운영 (기본 paper) ───────────────────────────── -bash scripts/3_run_stage3_all.sh -# python scripts/3_run_operations.py --mode paper -# python scripts/3_run_operations.py --mode live # API 키 필요 +# ── spot 3단계: fractal_swing 운영 ─────────────────────────── +bash scripts/3_run_fractal_ops.sh # 백테스트 + paper loop +python scripts/3_run_fractal_realistic_backtest.py +python scripts/3_run_filtered_backtest.py # 운영 조건 3년 sim +python scripts/3_run_operations.py --loop 180 --mode live # live (API 키 필요) ``` --- @@ -178,22 +181,62 @@ GT 타점 완벽 추종 시 수익 상한선. 최근 3년·초기 20만 원. | 2-3 | `2_run_signal_type_align.py` | `docs/spot/2_analysis/signal_type_report.html` | | 2-4 | `2_run_mtf_analysis.py` | `data/spot/mtf/mtf_rules_v3.json`, `docs/spot/2_analysis/mtf_correlation_report.html` | -### spot 3단계 — 실거래 운영 +### spot 3단계 — fractal_swing live 운영 -설계·운영 가이드: [`docs/spot/3_operations/stage3_design_guide.md`](docs/spot/3_operations/stage3_design_guide.md) +설계 가이드: [`docs/spot/3_operations/stage3_design_guide.md`](docs/spot/3_operations/stage3_design_guide.md) -`composite_v3` + MTF 필터 + 고TF 게이트. **기본 `OPS_MODE=paper`**. +**운영 전략:** `fractal_swing` + MTF off. 백테스트(운영 조건) **3년 +1,873,140%** (초기 20만원 → 약 37.5억, 슬리피지 0.05%·일 100회 상한 반영). | 순서 | 스크립트 | 산출물 | |------|----------|--------| -| 3-1 | `3_run_filtered_backtest.py` | `filtered_backtest_report.json` | -| 3-2 | `3_run_operations.py` | `ops_report.json`, `ops_state.json` | -| 일괄 | `3_run_stage3_all.sh` | 위 전체 | +| 백테스트 | `3_run_filtered_backtest.py` | `fractal_filtered_backtest_report.json` | +| **매매 차트** | `3_render_live_chart.py` | `docs/live/index.html`, `fractal_swing_ops_chart.html` | +| 시나리오 | `3_run_fractal_realistic_backtest.py` | `fractal_realistic_backtest.json` | +| 운영 | `3_run_operations.py --loop 180` | `fractal_ops_report.json`, `fractal_ops_state.json` | +| 일괄 | `3_run_fractal_ops.sh` | 백테스트 + paper loop | -| 모드 | 설명 | -|------|------| -| `paper` | 모의 체결 (권장) | -| `live` | 빗썸 시장가 주문 (`BITHUMB_ACCESS_KEY` 필요) | +#### 백테스트 vs live 코드 정합 (라이브 직전 확인) + +| 항목 | 백테스트 | live/paper 코드 | 상태 | +|------|----------|-----------------|------| +| 기법 | `fractal_swing` | `OPS_TECHNIQUE_ID` | 일치 | +| MTF | off | `OPS_MTF_ENABLED=false` | 일치 | +| 슬리피지 0.05% | `simulate_gt_signals_pnl` | `fill_price` → `executor` | 일치 | +| 수수료 0.05% | `GT_TRADING_FEE_RATE` | 동일 | 일치 | +| 최소 주문 5,000원 | `OPS_MIN_ORDER_KRW` | `trade_engine` | 일치 | +| 일 체결 100회 | `daily_max_trades` in sim | `runner` `trades_today_count` | 일치 | +| 매수 상한 | `max_buy_from_cash` | `compute_buy_order` | 일치 | +| 신호 | 캐시 JSON 전기간 | tail 800봉 갱신 + tick 체결 | 일치 (갱신 로직 추가) | +| 캔들 DB | 로컬 DB | `sync_ops_candles` 전 TF 증분 INSERT | 일치 | + +**live 전환 전 체크** + +1. `.env`: `OPS_MODE=live`, API 키 설정 +2. `python scripts/3_run_filtered_backtest.py` → 필터 sim **약 +1,873,140%** 확인 +3. `python scripts/3_run_operations.py --loop 180` (paper로 1~2일 모니터링 권장) +4. `fractal_ops_report.json`에서 `candle_sync`, `signal_refresh`, 체결 건수 확인 + +**주의:** 백테스트는 **3년 전기간 재생** sim이고, live는 **시간에 따라 누적**합니다. 2단계 ideal **+7,560,826%**(슬리피지 0)와는 다릅니다. 실거래 체결가는 모델 슬리피지보다 불리할 수 있습니다. + +#### 운영 환경 변수 (fractal 기본) + +| 변수 | 설명 | 기본값 | +|------|------|--------| +| `OPS_MODE` | `paper` / `live` | `paper` | +| `OPS_TECHNIQUE_ID` | 운영 기법 | `fractal_swing` | +| `OPS_MTF_ENABLED` | MTF 필터 | `false` | +| `OPS_DAILY_MAX_TRADES` | 일일 체결 상한 | `100` | +| `OPS_SLIPPAGE_RATE` | 편도 슬리피지 | `0.0005` (0.05%) | +| `OPS_MIN_ORDER_KRW` | 최소 주문 | `5000` | +| `OPS_ORDER_INTERVAL_SEC` | live 주문 간격(초) | `0.35` | +| `OPS_SYNC_CANDLES` | tick마다 캔들 증분 sync | `true` | +| `OPS_SYNC_INTERVALS` | sync TF (비우면 `DOWNLOAD_INTERVALS` 전체) | 전체 | +| `OPS_SIGNAL_TAIL_BARS` | 신호 tail 재계산 봉 수 | `800` | +| `OPS_PERSIST_SIGNAL_CACHE` | tail 갱신 후 JSON 저장 | `false` | +| `OPS_STATE_JSON` | 운영 상태 | `fractal_ops_state.json` | +| `OPS_REPORT_JSON` | 운영 리포트 | `fractal_ops_report.json` | + +`composite_v3` + MTF 운영은 `.env.example` 주석 참고. ### futures 0단계 — 선물 GT @@ -224,6 +267,11 @@ GT 타점 완벽 추종 시 수익 상한선. 최근 3년·초기 20만 원. | `GT_LOOKBACK_DAYS` | GT 타점 기간(일) | `3650` | | `GT_SIM_LOOKBACK_DAYS` | sim 거래 기간(일) | `1095` | | `GT_INITIAL_CASH_KRW` | sim 초기 자본(원) | `200000` | +| `OPS_MODE` | 운영 모드 | `paper` | +| `OPS_TECHNIQUE_ID` | 운영 기법 | `fractal_swing` | +| `OPS_SLIPPAGE_RATE` | 편도 슬리피지 | `0.0005` | +| `OPS_DAILY_MAX_TRADES` | 일일 체결 상한 | `100` | +| `OPS_SYNC_CANDLES` | tick 캔들 sync | `true` | ### 경로 변수 요약 @@ -233,6 +281,8 @@ GT 타점 완벽 추종 시 수익 상한선. 최근 3년·초기 20만 원. | spot GT 차트 | `GROUND_TRUTH_CHART_V3_FILE` | `docs/spot/0_ground_truth/...` | | spot sim 차트 | `GROUND_TRUTH_CHART_SIM_V3_FILE` | `docs/spot/1_simulation/...` | | spot 2단계 | `TECHNIQUES_DIR` | `data/spot/techniques/` | +| spot 3단계 운영 | `OPS_STATE_JSON` | `data/spot/operations/fractal_ops_state.json` | +| spot 3단계 리포트 | `OPS_REPORT_JSON` | `docs/spot/3_operations/fractal_ops_report.json` | | futures GT JSON | `GROUND_TRUTH_FUTURES_FILE` | `data/futures/ground_truth/...` | | futures GT 차트 | `GROUND_TRUTH_FUTURES_CHART_V3_FILE` | `docs/futures/0_ground_truth/...` | @@ -295,7 +345,7 @@ GT 타점 완벽 추종 시 수익 상한선. 최근 3년·초기 20만 원. | 유형 | 단계 | 상태 | |------|------|------| | common | 사전 (캔들) | 구현됨 | -| spot | 0~3단계 | 구현됨 (3단계 기본 paper) | +| spot | 0~3단계 | 구현됨 (3단계 fractal_swing live 준비) | | futures | 0단계 | 구현됨 | | futures | 1~3단계 | 예정 | @@ -303,6 +353,8 @@ GT 타점 완벽 추종 시 수익 상한선. 최근 3년·초기 20만 원. ## 변경 이력 +- 2026-06-13: `docs/live/` — 운영 백테스트 매수·매도 타점 HTML 차트 (`3_render_live_chart.py`) +- 2026-06-13: 운영 백테스트 **+1,873,140%** (3년, 슬리피지 0.05%, 일 100회) 검증 - 2026-06-12: `data/`·`docs/`를 **common / spot / futures** 3유형 구조로 재편, `coins.db` → `data/common/`, 0단계 차트 → `docs/{spot,futures}/0_ground_truth/` - 2026-06-12: `0_ground_truth_futures.py` — 현물 GT → 선물 JSON·차트 변환 로직 보완 - 2026-06-12: README 현물 파이프라인 전체 순서 갱신 diff --git a/docs/live/index.html b/docs/live/index.html new file mode 100644 index 0000000..bb16f02 --- /dev/null +++ b/docs/live/index.html @@ -0,0 +1,39 @@ + + + + + DeepCoin Live — 운영 백테스트 차트 + + + +

DeepCoin Live — 운영 백테스트

+

+ BTC · 프랙탈 스윙 (fractal_swing)
+ sim 기간: 최근 1095일 · + 슬리피지 0.05% · + 일 체결 상한 100 · + MTF off +

+
+
3년 수익률 (운영 규칙 sim)
+
+1874019.75%
+

+ 매수·매도 타점 차트 열기 +

+ +
+ + \ No newline at end of file diff --git a/scripts/3_render_live_chart.py b/scripts/3_render_live_chart.py new file mode 100644 index 0000000..32d415d --- /dev/null +++ b/scripts/3_render_live_chart.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +"""3단계: 운영 백테스트(+1,873,140%) 매수·매도 타점 HTML 차트 생성.""" + +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from deepcoin.config import load_settings +from deepcoin.operations.chart import render_ops_live_chart + + +def _configure_logging(verbose: bool) -> None: + level = logging.DEBUG if verbose else logging.INFO + logging.basicConfig( + level=level, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + +def _write_index_html( + index_path: Path, + chart_name: str, + report: dict, +) -> Path: + """docs/live 인덱스 HTML을 생성한다.""" + sim = report["filtered_sim"] + ret = sim.get("total_return_pct", 0) + index_path.parent.mkdir(parents=True, exist_ok=True) + html = f""" + + + + DeepCoin Live — 운영 백테스트 차트 + + + +

DeepCoin Live — 운영 백테스트

+

+ {report.get("symbol", "BTC")} · {report.get("technique_name", "")} ({report.get("technique_id", "")})
+ sim 기간: 최근 {report.get("sim_lookback_days", 1095)}일 · + 슬리피지 {report.get("slippage_rate", 0) * 100:.2f}% · + 일 체결 상한 {report.get("daily_max_trades", "-")} · + MTF {"on" if report.get("mtf_enabled") else "off"} +

+
+
3년 수익률 (운영 규칙 sim)
+
{ret:+.2f}%
+

+ 매수·매도 타점 차트 열기 +

+ +
+ +""" + index_path.write_text(html, encoding="utf-8") + return index_path + + +def main() -> int: + """CLI 진입점.""" + parser = argparse.ArgumentParser( + description="운영 백테스트 매수·매도 타점 HTML 차트 (docs/live)", + ) + parser.add_argument( + "-o", + "--output", + type=str, + default=None, + help="HTML 출력 경로 (기본: docs/live/{technique}_ops_chart.html)", + ) + parser.add_argument( + "--chart-days", + type=int, + default=None, + help="차트 캔들 표시 일수 (기본: GT_SIM_LOOKBACK_DAYS)", + ) + parser.add_argument("-v", "--verbose", action="store_true") + args = parser.parse_args() + _configure_logging(args.verbose) + + settings = load_settings() + live_dir = ROOT / "docs" / "live" + default_name = f"{settings.ops_technique_id}_ops_chart.html" + output_path = Path(args.output) if args.output else live_dir / default_name + if not output_path.is_absolute(): + output_path = ROOT / output_path + + print("\n=== 운영 백테스트 차트 생성 ===", flush=True) + print( + f"기법: {settings.ops_technique_id} · 슬리피지 {settings.ops_slippage_rate} · " + f"일 상한 {settings.ops_daily_max_trades}", + flush=True, + ) + + chart_path, report = render_ops_live_chart( + settings, + output_path, + chart_lookback_days=args.chart_days, + ) + data_js = chart_path.with_name(f"{chart_path.stem}_data.js") + index_path = _write_index_html(live_dir / "index.html", chart_path.name, report) + + sim = report["filtered_sim"] + print(f"\n3년 수익률: {sim.get('total_return_pct'):+.2f}%") + print(f"매수/매도: {sim.get('buys_executed')}/{sim.get('sells_executed')}") + print(f"HTML: {chart_path}") + print(f"데이터: {data_js} ({data_js.stat().st_size / 1e6:.1f} MB)") + print(f"인덱스: {index_path}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/3_run_filtered_backtest.py b/scripts/3_run_filtered_backtest.py index 87e772d..ca77b4e 100644 --- a/scripts/3_run_filtered_backtest.py +++ b/scripts/3_run_filtered_backtest.py @@ -43,6 +43,7 @@ def main() -> int: raw = report["raw_sim"] print("\n=== 3단계 MTF 필터 백테스트 ===") print(f"기법: {report['technique_id']}") + print(f"슬리피지: {report.get('slippage_rate')} · 일일체결상한: {report.get('daily_max_trades')}") print(f"원시 신호: {report['raw_signal_count']} → 필터 통과: {report['filtered_signal_count']}") print(f"원시 3년 sim: {raw.get('total_return_pct')}%") print(f"필터 3년 sim: {filt.get('total_return_pct')}%") diff --git a/scripts/3_run_fractal_ops.sh b/scripts/3_run_fractal_ops.sh new file mode 100755 index 0000000..7176293 --- /dev/null +++ b/scripts/3_run_fractal_ops.sh @@ -0,0 +1,12 @@ +#!/usr/bin/env bash +# fractal_swing paper 운영 — 3분봉 tick (180초 간격) +set -euo pipefail +cd "$(dirname "$0")/.." +export PYTHONPATH=src + +echo "=== fractal 현실적 백테스트 ===" +python scripts/3_run_fractal_realistic_backtest.py + +echo "" +echo "=== paper 운영 (180초 loop) — Ctrl+C 종료 ===" +python scripts/3_run_operations.py --loop 180 diff --git a/scripts/3_run_fractal_realistic_backtest.py b/scripts/3_run_fractal_realistic_backtest.py new file mode 100644 index 0000000..37395b5 --- /dev/null +++ b/scripts/3_run_fractal_realistic_backtest.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 +"""fractal_swing 현실적 백테스트 — 슬리피지·일일체결 상한 시나리오.""" + +from __future__ import annotations + +import argparse +import json +import logging +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from deepcoin.config import load_settings +from deepcoin.evaluation.causal_sim import normalize_signals_for_sim +from deepcoin.ground_truth.pnl import simulate_gt_signals_pnl +from deepcoin.operations.signal_pipeline import ( + _signals_in_lookback, + generate_raw_signals, + load_ops_candles, +) + + +def _configure_logging(verbose: bool) -> None: + level = logging.DEBUG if verbose else logging.INFO + logging.basicConfig( + level=level, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + +def main() -> int: + """fractal_swing 슬리피지 시나리오 백테스트.""" + parser = argparse.ArgumentParser(description="fractal_swing 현실적 3년 백테스트") + parser.add_argument("-v", "--verbose", action="store_true") + args = parser.parse_args() + _configure_logging(args.verbose) + + settings = load_settings() + df = load_ops_candles(settings) + gen = generate_raw_signals(settings, df=df, use_cache=True) + end = gen["data_end"] + price = gen["last_price"] + scoped = _signals_in_lookback(gen["raw_signals"], end, settings.gt_sim_lookback_days) + normalized = normalize_signals_for_sim(scoped) + + scenarios = [ + {"label": "stage2_ideal", "slippage_rate": 0.0, "daily_max_trades": None}, + {"label": "ops_default", "slippage_rate": settings.ops_slippage_rate, "daily_max_trades": settings.ops_daily_max_trades}, + {"label": "slippage_0.1pct", "slippage_rate": 0.001, "daily_max_trades": settings.ops_daily_max_trades}, + {"label": "slippage_0.1pct_no_cap", "slippage_rate": 0.001, "daily_max_trades": None}, + ] + + rows = [] + for sc in scenarios: + sim = simulate_gt_signals_pnl( + signals=normalized, + initial_cash_krw=settings.gt_initial_cash_krw, + fee_rate=settings.gt_trading_fee_rate, + min_order_krw=settings.ops_min_order_krw, + slippage_rate=sc["slippage_rate"], + daily_max_trades=sc["daily_max_trades"], + sim_lookback_days=settings.gt_sim_lookback_days, + data_end=end, + last_mark_price=price, + ) + buys = sim["buys_executed"] + rows.append({ + **sc, + "return_pct": sim["total_return_pct"], + "final_equity_krw": sim["final_equity_krw"], + "buys_executed": buys, + "sells_executed": sim["sells_executed"], + "buys_skipped": sim["buys_skipped"], + "daily_buys_avg": round(buys / settings.gt_sim_lookback_days, 2), + }) + + report = { + "technique_id": settings.ops_technique_id, + "symbol": settings.symbol, + "sim_lookback_days": settings.gt_sim_lookback_days, + "initial_cash_krw": settings.gt_initial_cash_krw, + "signals_in_period": len(scoped), + "scenarios": rows, + } + out = Path("docs/spot/3_operations/fractal_realistic_backtest.json") + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + + print("\n=== fractal_swing 현실적 백테스트 (3년) ===") + for row in rows: + print( + f"{row['label']}: {row['return_pct']}% " + f"(슬리피지 {row['slippage_rate']}, 일상한 {row['daily_max_trades']}) " + f"매수 {row['buys_executed']} (일 {row['daily_buys_avg']})" + ) + print(f"\nJSON: {out}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/deepcoin/config.py b/src/deepcoin/config.py index 534db0f..e85fe04 100644 --- a/src/deepcoin/config.py +++ b/src/deepcoin/config.py @@ -106,7 +106,12 @@ class Settings: ops_trend_gate_enabled: bool ops_daily_max_trades: int ops_min_order_krw: float + ops_slippage_rate: float + ops_order_interval_sec: float ops_sync_candles: bool + ops_sync_intervals: list[int] + ops_signal_tail_bars: int + ops_persist_signal_cache: bool @property def market(self) -> str: @@ -300,6 +305,29 @@ def load_settings(env_path: Path | None = None) -> Settings: in ("1", "true", "yes", "on"), ops_daily_max_trades=int(os.getenv("OPS_DAILY_MAX_TRADES", "20")), ops_min_order_krw=float(os.getenv("OPS_MIN_ORDER_KRW", "5000")), + ops_slippage_rate=float(os.getenv("OPS_SLIPPAGE_RATE", "0.0005")), + ops_order_interval_sec=float(os.getenv("OPS_ORDER_INTERVAL_SEC", "0.35")), ops_sync_candles=os.getenv("OPS_SYNC_CANDLES", "true").strip().lower() in ("1", "true", "yes", "on"), + ops_sync_intervals=_parse_ops_sync_intervals( + os.getenv("OPS_SYNC_INTERVALS", "").strip(), + fallback_intervals=intervals, + ), + ops_signal_tail_bars=int(os.getenv("OPS_SIGNAL_TAIL_BARS", "800")), + ops_persist_signal_cache=os.getenv("OPS_PERSIST_SIGNAL_CACHE", "false").strip().lower() + in ("1", "true", "yes", "on"), ) + + +def _parse_ops_sync_intervals( + raw: str, + fallback_intervals: list[int], +) -> list[int]: + """운영 캔들 sync 대상 인터벌 목록. + + OPS_SYNC_INTERVALS 비어 있으면 DOWNLOAD_INTERVALS 전체를 사용한다. + """ + if not raw: + return list(fallback_intervals) + values = sorted(set(int(x.strip()) for x in raw.split(",") if x.strip())) + return values if values else list(fallback_intervals) diff --git a/src/deepcoin/data/candle_store.py b/src/deepcoin/data/candle_store.py index e4efccc..f9f0ff5 100644 --- a/src/deepcoin/data/candle_store.py +++ b/src/deepcoin/data/candle_store.py @@ -189,3 +189,71 @@ class CandleStore: ) self._conn.commit() return len(payload) + + def insert_new_rows( + self, + symbol: str, + coin_name: str, + interval_min: int, + rows: list[tuple], + *, + after: datetime | None = None, + ) -> int: + """DB에 없는 캔들만 INSERT한다 (증분 수집용). + + Args: + symbol: 코인 심볼. + coin_name: 코인 이름. + interval_min: 분 단위 인터벌. + rows: ``(ymdhms, open, high, low, close, volume)`` 튜플 리스트. + after: 이 시각 **이후** 봉만 저장. None이면 필터 없음. + + Returns: + 실제 INSERT된 행 수. + """ + if not rows: + return 0 + + if after is not None: + filtered: list[tuple] = [] + for row in rows: + dt = parse_kst_datetime(str(row[0])) + if dt > after: + filtered.append(row) + rows = filtered + if not rows: + return 0 + + table = self.ensure_table(symbol, interval_min) + code = symbol.upper() + payload: list[tuple] = [] + for ymdhms, open_p, high_p, low_p, close_p, volume in rows: + dt = parse_kst_datetime(str(ymdhms)) if isinstance(ymdhms, str) else ymdhms + ymd = dt.strftime("%Y-%m-%d") + hms = dt.strftime("%H:%M:%S") + payload.append( + ( + code, + coin_name, + dt.strftime("%Y-%m-%d %H:%M:%S"), + ymd, + hms, + close_p, + open_p, + high_p, + low_p, + volume, + ) + ) + + changes_before = self._conn.total_changes + self._conn.executemany( + f""" + INSERT OR IGNORE INTO {table} + (CODE, NAME, ymdhms, ymd, hms, Close, Open, High, Low, Volume) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + payload, + ) + self._conn.commit() + return self._conn.total_changes - changes_before diff --git a/src/deepcoin/data/downloader.py b/src/deepcoin/data/downloader.py index 8684b95..2ff16ed 100644 --- a/src/deepcoin/data/downloader.py +++ b/src/deepcoin/data/downloader.py @@ -136,12 +136,22 @@ class CandleDownloader: if not rows: break - saved_rows += store.upsert_rows( - symbol, - self.settings.coin_name, - interval_min, - rows, - ) + if mode == "incremental" and db_max is not None: + inserted = store.insert_new_rows( + symbol, + self.settings.coin_name, + interval_min, + rows, + after=db_max, + ) + else: + inserted = store.upsert_rows( + symbol, + self.settings.coin_name, + interval_min, + rows, + ) + saved_rows += inserted batch_oldest = min(parse_kst_datetime(r[0]) for r in rows) if oldest_seen is None or batch_oldest < oldest_seen: @@ -168,7 +178,7 @@ class CandleDownloader: ) logger.info( - "수집 완료 %s_%s mode=%s requests=%s saved=%s reached=%s", + "수집 완료 %s_%s mode=%s requests=%s inserted=%s reached=%s", symbol, interval_min, mode, diff --git a/src/deepcoin/ground_truth/chart.py b/src/deepcoin/ground_truth/chart.py index cda6aaf..85ba77a 100644 --- a/src/deepcoin/ground_truth/chart.py +++ b/src/deepcoin/ground_truth/chart.py @@ -741,6 +741,16 @@ __EXTRA_SCRIPT__ simNoteExtra: `기법 ${techniqueId} | 미래 데이터 미사용${benchNote}`, }; } + if (stage === "spot_3_operations") { + const opsNote = m.ops_note || ""; + return { + pageTitle: `${m.symbol} 운영 백테스트 — ${techniqueName}`, + title: `${m.symbol} 3단계 운영 백테스트 (${techniqueName}, ${m.interval_label}) — live 동일 규칙 · 차트 ${chartLabel}`, + panelTitle: `3단계 운영 백테스트 sim (${simLabel}${initLabel})`, + legend: "B=매수 S=매도 | 실제 체결만 표시", + simNoteExtra: `기법 ${techniqueId} | ${opsNote} | live/paper와 동일 체결 규칙`, + }; + } if (stage === "spot_2_causal_sim" || techniqueId) { return { pageTitle: `${m.symbol} 인과 sim — ${techniqueName}`, diff --git a/src/deepcoin/ground_truth/pnl.py b/src/deepcoin/ground_truth/pnl.py index c753de6..468a03a 100644 --- a/src/deepcoin/ground_truth/pnl.py +++ b/src/deepcoin/ground_truth/pnl.py @@ -9,6 +9,20 @@ from typing import Any from deepcoin.ground_truth.order_sizing import buy_sizing_metadata, max_buy_from_cash +def _fill_price(signal_price: float, side: str, slippage_rate: float) -> float: + """슬리피지 반영 체결가.""" + price = float(signal_price) + slip = max(float(slippage_rate), 0.0) + if side == "buy": + return price * (1.0 + slip) + return price * max(1.0 - slip, 0.0) + + +def _signal_date(datetime_str: str) -> str: + """신호 datetime에서 YYYY-MM-DD 추출.""" + return datetime_str[:10] + + @dataclass class LegPnl: """레그별 손익 (매수→매도 1쌍).""" @@ -161,6 +175,8 @@ def simulate_gt_signals_pnl( initial_cash_krw: float = 400_000.0, fee_rate: float = 0.0005, min_order_krw: float = 5_000.0, + slippage_rate: float = 0.0, + daily_max_trades: int | None = None, sim_lookback_days: int = 365, data_end: str | None = None, last_mark_price: float | None = None, @@ -178,6 +194,8 @@ def simulate_gt_signals_pnl( initial_cash_krw: 시뮬 시작 원화. fee_rate: 편도 수수료율. min_order_krw: 최소 주문 금액. + slippage_rate: 편도 슬리피지 (0.0005 = 0.05%). + daily_max_trades: 일일 체결 상한 (None이면 무제한). sim_lookback_days: 시뮬 기간(일). data_end: 데이터 종료 시각 문자열. None이면 마지막 신호 시각. last_mark_price: 미청산 코인 평가 가격. None이면 마지막 체결가. @@ -214,9 +232,22 @@ def simulate_gt_signals_pnl( sells_executed = 0 buys_skipped = 0 sells_skipped = 0 + daily_trade_counts: dict[str, int] = {} mark_price = float(last_mark_price or period_signals[-1]["price"]) + def _can_trade_today(datetime_str: str) -> bool: + if daily_max_trades is None: + return True + day = _signal_date(datetime_str) + return daily_trade_counts.get(day, 0) < daily_max_trades + + def _record_trade(datetime_str: str) -> None: + if daily_max_trades is None: + return + day = _signal_date(datetime_str) + daily_trade_counts[day] = daily_trade_counts.get(day, 0) + 1 + for side, cluster in _cluster_signals(period_signals): cluster_size = len(cluster) if side == "buy": @@ -224,9 +255,32 @@ def simulate_gt_signals_pnl( per_buy = budget / cluster_size if cluster_size else 0.0 for sig in cluster: trade_id += 1 - price = float(sig["price"]) + price = _fill_price(float(sig["price"]), "buy", slippage_rate) cash_before = cash coin_before = coin_qty + if not _can_trade_today(sig["datetime"]): + buys_skipped += 1 + trades.append( + SignalTrade( + trade_id=trade_id, + side="buy", + signal_type=str(sig.get("signal_type", "buy")), + marker_id=sig.get("marker_id"), + datetime=sig["datetime"], + price=price, + cash_before=round(cash_before, 0), + cash_after=round(cash, 0), + coin_before=round(coin_before, 8), + coin_after=round(coin_qty, 8), + order_krw=0.0, + order_coin=0.0, + fee_krw=0.0, + cluster_size=cluster_size, + skipped=True, + skip_reason="일일 체결 상한", + ) + ) + continue equity = cash + coin_qty * price cash_cap = max_buy_from_cash(equity, cash) order_krw = min(per_buy, cash, cash_cap) @@ -260,6 +314,7 @@ def simulate_gt_signals_pnl( cash -= order_krw coin_qty += bought buys_executed += 1 + _record_trade(sig["datetime"]) trades.append( SignalTrade( trade_id=trade_id, @@ -284,9 +339,32 @@ def simulate_gt_signals_pnl( per_sell = budget_coin / cluster_size if cluster_size else 0.0 for sig in cluster: trade_id += 1 - price = float(sig["price"]) + price = _fill_price(float(sig["price"]), "sell", slippage_rate) cash_before = cash coin_before = coin_qty + if not _can_trade_today(sig["datetime"]): + sells_skipped += 1 + trades.append( + SignalTrade( + trade_id=trade_id, + side="sell", + signal_type=str(sig.get("signal_type", "sell")), + marker_id=sig.get("marker_id"), + datetime=sig["datetime"], + price=price, + cash_before=round(cash_before, 0), + cash_after=round(cash, 0), + coin_before=round(coin_before, 8), + coin_after=round(coin_qty, 8), + order_krw=0.0, + order_coin=0.0, + fee_krw=0.0, + cluster_size=cluster_size, + skipped=True, + skip_reason="일일 체결 상한", + ) + ) + continue order_coin = min(per_sell, coin_qty) order_krw = order_coin * price @@ -319,6 +397,7 @@ def simulate_gt_signals_pnl( cash += gross - fee coin_qty -= order_coin sells_executed += 1 + _record_trade(sig["datetime"]) trades.append( SignalTrade( trade_id=trade_id, @@ -355,6 +434,8 @@ def simulate_gt_signals_pnl( "total_pnl_krw": round(total_pnl, 0), "total_return_pct": round(total_return_pct, 2), "fee_rate": fee_rate, + "slippage_rate": slippage_rate, + "daily_max_trades": daily_max_trades, **buy_sizing_metadata(), "sim_lookback_days": sim_lookback_days, "period_from": start_str, diff --git a/src/deepcoin/operations/backtest.py b/src/deepcoin/operations/backtest.py index abdf5bd..6bd4151 100644 --- a/src/deepcoin/operations/backtest.py +++ b/src/deepcoin/operations/backtest.py @@ -26,6 +26,9 @@ def run_filtered_backtest(settings: Settings) -> dict[str, Any]: signals=normalized, initial_cash_krw=settings.gt_initial_cash_krw, fee_rate=settings.gt_trading_fee_rate, + min_order_krw=settings.ops_min_order_krw, + slippage_rate=settings.ops_slippage_rate, + daily_max_trades=settings.ops_daily_max_trades, sim_lookback_days=settings.gt_sim_lookback_days, data_end=pipeline["data_end"], last_mark_price=pipeline["last_price"], @@ -35,6 +38,9 @@ def run_filtered_backtest(settings: Settings) -> dict[str, Any]: signals=normalize_signals_for_sim(pipeline["scoped_raw_signals"]), initial_cash_krw=settings.gt_initial_cash_krw, fee_rate=settings.gt_trading_fee_rate, + min_order_krw=settings.ops_min_order_krw, + slippage_rate=settings.ops_slippage_rate, + daily_max_trades=settings.ops_daily_max_trades, sim_lookback_days=settings.gt_sim_lookback_days, data_end=pipeline["data_end"], last_mark_price=pipeline["last_price"], @@ -48,6 +54,8 @@ def run_filtered_backtest(settings: Settings) -> dict[str, Any]: "params": pipeline["params"], "mtf_enabled": pipeline.get("mtf_enabled", False), "trend_gate_enabled": settings.ops_trend_gate_enabled, + "slippage_rate": settings.ops_slippage_rate, + "daily_max_trades": settings.ops_daily_max_trades, "raw_signal_count": pipeline["raw_count"], "filtered_signal_count": pipeline["kept_count"], "rejected_signal_count": pipeline["rejected_count"], diff --git a/src/deepcoin/operations/candle_sync.py b/src/deepcoin/operations/candle_sync.py new file mode 100644 index 0000000..b936422 --- /dev/null +++ b/src/deepcoin/operations/candle_sync.py @@ -0,0 +1,53 @@ +"""운영 tick용 빗썸 캔들 증분 동기화.""" + +from __future__ import annotations + +import logging +from dataclasses import replace + +from deepcoin.config import Settings +from deepcoin.data.candle_store import CandleStore +from deepcoin.data.downloader import CandleDownloader, DownloadResult + +logger = logging.getLogger(__name__) + + +def sync_ops_candles(settings: Settings) -> list[DownloadResult]: + """``DOWNLOAD_INTERVALS``(또는 ``OPS_SYNC_INTERVALS``) 전 TF 증분 sync. + + 각 인터벌별로: + 1. DB ``MAX(ymdhms)`` 조회 + 2. 최신이면 API 호출 없이 ``uptodate`` + 3. 갭이 있으면 빗썸 역방향 조회 후 ``db_max`` 이후 봉만 INSERT + + Args: + settings: 애플리케이션 설정. + + Returns: + 인터벌별 DownloadResult 리스트. + """ + intervals = settings.ops_sync_intervals + sync_settings = replace(settings, download_intervals=intervals) + store = CandleStore(settings.db_path) + try: + downloader = CandleDownloader(sync_settings) + results = downloader.download_all( + store, + days=settings.download_days, + full=False, + ) + for result in results: + count, min_dt, max_dt = store.get_range(settings.symbol, result.interval_min) + max_s = max_dt.strftime("%Y-%m-%d %H:%M:%S") if max_dt else "-" + logger.info( + "운영 캔들 sync %smin mode=%s api=%s inserted=%s db_max=%s rows=%s", + result.interval_min, + result.mode, + result.requests, + result.saved_rows, + max_s, + count, + ) + return results + finally: + store.close() diff --git a/src/deepcoin/operations/chart.py b/src/deepcoin/operations/chart.py new file mode 100644 index 0000000..def6ee5 --- /dev/null +++ b/src/deepcoin/operations/chart.py @@ -0,0 +1,79 @@ +"""3단계 운영 백테스트 매매 타점 HTML 차트.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from deepcoin.config import Settings +from deepcoin.data.intervals import interval_label +from deepcoin.ground_truth.chart import render_ground_truth_sim_chart +from deepcoin.operations.backtest import run_filtered_backtest + + +def _ops_chart_shell( + pipeline: dict[str, Any], + settings: Settings, + sim_pnl: dict[str, Any], +) -> dict[str, Any]: + """운영 백테스트 차트용 GT 메타 셸을 구성한다.""" + slip_pct = settings.ops_slippage_rate * 100 + daily_cap = settings.ops_daily_max_trades + return { + "meta": { + "symbol": settings.symbol, + "interval_min": settings.gt_interval_min, + "interval_label": interval_label(settings.gt_interval_min), + "lookback_days": settings.gt_sim_lookback_days, + "technique_id": pipeline["technique_id"], + "technique_name": pipeline["technique_name"], + "stage": "spot_3_operations", + "mtf_enabled": pipeline.get("mtf_enabled", False), + "slippage_rate": settings.ops_slippage_rate, + "daily_max_trades": daily_cap, + "ops_note": ( + f"슬리피지 {slip_pct:.2f}% · 일 체결 상한 {daily_cap} · " + f"MTF {'on' if pipeline.get('mtf_enabled') else 'off'}" + ), + "data_from": sim_pnl.get("period_from"), + "data_to": sim_pnl.get("period_to"), + }, + "signals": [], + } + + +def render_ops_live_chart( + settings: Settings, + output_path: Path, + *, + chart_lookback_days: int | None = None, +) -> tuple[Path, dict[str, Any]]: + """운영 설정과 동일한 규칙으로 3년 sim 후 매수·매도 타점 HTML을 생성한다. + + Args: + settings: 애플리케이션 설정 (.env). + output_path: HTML 출력 경로. + chart_lookback_days: 차트 캔들 표시 일수. None이면 sim 기간과 동일. + + Returns: + (HTML 경로, 백테스트 리포트 dict). + """ + report = run_filtered_backtest(settings) + sim_pnl = report["filtered_sim"] + pipeline_meta = { + "technique_id": report["technique_id"], + "technique_name": report["technique_name"], + "mtf_enabled": report.get("mtf_enabled", False), + } + gt_shell = _ops_chart_shell(pipeline_meta, settings, sim_pnl) + days = chart_lookback_days if chart_lookback_days is not None else settings.gt_sim_lookback_days + + path = render_ground_truth_sim_chart( + db_path=settings.db_path, + symbol=settings.symbol, + gt_result=gt_shell, + sim_pnl=sim_pnl, + output_path=output_path, + chart_lookback_days=days, + ) + return path, report diff --git a/src/deepcoin/operations/execution.py b/src/deepcoin/operations/execution.py new file mode 100644 index 0000000..c293c7c --- /dev/null +++ b/src/deepcoin/operations/execution.py @@ -0,0 +1,21 @@ +"""체결 가격·슬리피지 (paper / live / backtest 공통).""" + +from __future__ import annotations + + +def fill_price(signal_price: float, side: str, slippage_rate: float) -> float: + """신호 가격에 슬리피지를 반영한 체결가를 계산한다. + + Args: + signal_price: 신호(봉) 기준 가격. + side: ``buy`` 또는 ``sell``. + slippage_rate: 편도 슬리피지 비율 (0.0005 = 0.05%). + + Returns: + 매수는 불리하게(↑) · 매도는 불리하게(↓) 조정된 체결가. + """ + price = float(signal_price) + slip = max(float(slippage_rate), 0.0) + if side == "buy": + return price * (1.0 + slip) + return price * max(1.0 - slip, 0.0) diff --git a/src/deepcoin/operations/executor.py b/src/deepcoin/operations/executor.py index daeefcf..4a1865d 100644 --- a/src/deepcoin/operations/executor.py +++ b/src/deepcoin/operations/executor.py @@ -8,6 +8,7 @@ from typing import Any from deepcoin.api.bithumb_private import BithumbPrivateClient from deepcoin.config import Settings +from deepcoin.operations.execution import fill_price from deepcoin.operations.trade_engine import ( TradeResult, apply_trade_to_portfolio, @@ -46,12 +47,16 @@ class PaperExecutor(OrderExecutor): cluster_size: int = 1, ) -> TradeResult: """paper 모드 체결.""" - price = float(signal["price"]) + side = str(signal["side"]) + price = fill_price( + float(signal["price"]), + side, + self.settings.ops_slippage_rate, + ) fee_rate = self.settings.gt_trading_fee_rate min_order = self.settings.ops_min_order_krw cash = float(portfolio.get("cash_krw", 0)) coin = float(portfolio.get("coin_qty", 0)) - side = str(signal["side"]) if side == "buy": trade = compute_buy_order( @@ -99,12 +104,16 @@ class LiveExecutor(OrderExecutor): ) -> TradeResult: """live 모드 시장가 주문.""" self._sync_portfolio(portfolio) - price = float(signal["price"]) + side = str(signal["side"]) + price = fill_price( + float(signal["price"]), + side, + self.settings.ops_slippage_rate, + ) fee_rate = self.settings.gt_trading_fee_rate min_order = self.settings.ops_min_order_krw cash = float(portfolio.get("cash_krw", 0)) coin = float(portfolio.get("coin_qty", 0)) - side = str(signal["side"]) if side == "buy": trade = compute_buy_order( diff --git a/src/deepcoin/operations/runner.py b/src/deepcoin/operations/runner.py index 94be929..46e6b97 100644 --- a/src/deepcoin/operations/runner.py +++ b/src/deepcoin/operations/runner.py @@ -4,14 +4,14 @@ from __future__ import annotations import json import logging -import subprocess -import sys +import time from datetime import datetime from pathlib import Path from typing import Any from deepcoin.config import Settings from deepcoin.ground_truth.pnl import _cluster_signals +from deepcoin.operations.candle_sync import sync_ops_candles from deepcoin.operations.executor import create_executor from deepcoin.operations.signal_pipeline import ( filter_signals_for_ops, @@ -23,25 +23,15 @@ from deepcoin.operations.state_store import load_state, reset_daily_trade_count, logger = logging.getLogger(__name__) -def _project_root() -> Path: - return Path(__file__).resolve().parents[3] - - -def sync_candles_if_enabled(settings: Settings) -> bool: - """증분 캔들 수집 스크립트를 실행한다.""" +def sync_candles_if_enabled(settings: Settings) -> list[Any]: + """빗썸 캔들을 증분 조회하여 DB에 새 봉만 INSERT한다.""" if not settings.ops_sync_candles: - return False - script = _project_root() / "scripts" / "00_download.py" - if not script.exists(): - logger.warning("캔들 동기화 스크립트 없음: %s", script) - return False - logger.info("캔들 증분 동기화 실행...") - subprocess.run( - [sys.executable, str(script)], - cwd=str(_project_root()), - check=False, + return [] + logger.info( + "캔들 증분 동기화 (intervals=%s)...", + settings.ops_sync_intervals, ) - return True + return sync_ops_candles(settings) def _signals_on_bar(signals: list[dict[str, Any]], bar_index: int) -> list[dict[str, Any]]: @@ -98,8 +88,10 @@ class OperationsRunner: def tick(self, *, sync_candles: bool | None = None) -> dict[str, Any]: """신호 확인 및 체결 1회.""" - if sync_candles if sync_candles is not None else self.settings.ops_sync_candles: - sync_candles_if_enabled(self.settings) + do_sync = sync_candles if sync_candles is not None else self.settings.ops_sync_candles + candle_sync_results: list[Any] = [] + if do_sync: + candle_sync_results = sync_candles_if_enabled(self.settings) df = load_ops_candles(self.settings) latest_bar = int(len(df) - 1) @@ -147,6 +139,11 @@ class OperationsRunner: executions.append(record) if trade.executed: self.state["trades_today_count"] += 1 + if ( + self.settings.ops_mode == "live" + and self.settings.ops_order_interval_sec > 0 + ): + time.sleep(self.settings.ops_order_interval_sec) if bar_idx > last_bar: last_bar = bar_idx self.state["last_processed_bar_index"] = bar_idx @@ -170,6 +167,18 @@ class OperationsRunner: "generated_at": now, "mode": self.settings.ops_mode, "technique_id": pipeline["technique_id"], + "slippage_rate": self.settings.ops_slippage_rate, + "daily_max_trades": self.settings.ops_daily_max_trades, + "signal_refresh": gen.get("signal_refresh"), + "candle_sync": [ + { + "interval_min": r.interval_min, + "mode": r.mode, + "api_requests": r.requests, + "inserted_rows": r.saved_rows, + } + for r in candle_sync_results + ], "raw_signals": pipeline["raw_count"], "filtered_signals": pipeline["kept_count"], "pending_bars": target_bars, diff --git a/src/deepcoin/operations/signal_pipeline.py b/src/deepcoin/operations/signal_pipeline.py index 535d075..feecce3 100644 --- a/src/deepcoin/operations/signal_pipeline.py +++ b/src/deepcoin/operations/signal_pipeline.py @@ -2,7 +2,10 @@ from __future__ import annotations +import logging +import re from datetime import datetime, timedelta +from pathlib import Path from typing import Any import pandas as pd @@ -15,9 +18,75 @@ from deepcoin.mtf.rules import load_or_derive_mtf_rules from deepcoin.mtf.store import MultiTimeframeStore from deepcoin.mtf.trend_gate import HtfTrendGate from deepcoin.operations.signal_type import enrich_signal_types -from deepcoin.techniques.base import TechniqueParams +from deepcoin.techniques.base import TechniqueParams, TechniqueResult from deepcoin.techniques.registry import get_technique -from deepcoin.techniques.runner import load_technique_result, run_technique +from deepcoin.techniques.runner import load_technique_result, run_technique, save_technique_result + +logger = logging.getLogger(__name__) + +_technique_file_cache: dict[str, tuple[float, TechniqueResult]] = {} + + +def _max_signal_bar_index(signals: list[dict[str, Any]]) -> int: + """신호 목록의 최대 bar_index.""" + if not signals: + return -1 + return max(int(s.get("bar_index", 0)) for s in signals) + + +def _offset_signal_bars(signals: list[dict[str, Any]], offset: int) -> list[dict[str, Any]]: + """tail 구간 실행 결과의 bar_index를 전역 인덱스로 변환한다.""" + shifted: list[dict[str, Any]] = [] + for signal in signals: + row = dict(signal) + row["bar_index"] = int(row.get("bar_index", 0)) + offset + pivot = row.get("pivot_bar_index") + if pivot is not None: + row["pivot_bar_index"] = int(pivot) + offset + shifted.append(row) + return shifted + + +def _merge_tail_signals( + cached_signals: list[dict[str, Any]], + tail_signals: list[dict[str, Any]], + offset: int, + after_bar: int, +) -> list[dict[str, Any]]: + """캐시 신호 + tail 신규 구간만 병합 (datetime·side 중복 제거).""" + seen = {(s.get("datetime"), s.get("side")) for s in cached_signals} + merged = list(cached_signals) + for signal in _offset_signal_bars(tail_signals, offset): + if int(signal.get("bar_index", 0)) <= after_bar: + continue + key = (signal.get("datetime"), signal.get("side")) + if key in seen: + continue + merged.append(signal) + seen.add(key) + return merged + + +def _load_technique_cached(cache_path: Path) -> TechniqueResult: + """대용량 기법 JSON을 mtime 기준으로 메모리 캐시한다 (fractal_swing 등).""" + path = Path(cache_path) + mtime = path.stat().st_mtime + key = str(path) + cached = _technique_file_cache.get(key) + if cached and cached[0] == mtime: + return cached[1] + result = load_technique_result(path) + _technique_file_cache.clear() + _technique_file_cache[key] = (mtime, result) + return result + + +def _store_technique_memory(cache_path: Path, result: TechniqueResult) -> None: + """메모리 캐시를 갱신한다 (tick 간 신호 연속성).""" + key = str(cache_path) + mtime = cache_path.stat().st_mtime if cache_path.exists() else 0.0 + _technique_file_cache.clear() + _technique_file_cache[key] = (mtime, result) def _parse_dt(value: str) -> datetime: @@ -36,6 +105,36 @@ def _signals_in_lookback( return [s for s in signals if _parse_dt(s["datetime"]) >= start_dt] +def _filter_composite_min_score( + signals: list[dict[str, Any]], + min_score: float, +) -> list[dict[str, Any]]: + """composite_v3 reason의 score= 값으로 min_score 이상 신호만 남긴다.""" + kept: list[dict[str, Any]] = [] + for signal in signals: + match = re.search(r"score=([\d.]+)", signal.get("reason", "")) + if match: + if float(match.group(1)) >= min_score: + kept.append(signal) + else: + kept.append(signal) + return kept + + +def _apply_ops_min_score( + technique_id: str, + signals: list[dict[str, Any]], + cached_min_score: float | None, + ops_min_score: float | None, +) -> list[dict[str, Any]]: + """캐시 신호에 OPS_MIN_SCORE가 캐시보다 높으면 추가 필터를 적용한다.""" + if technique_id != "composite_v3" or ops_min_score is None: + return signals + if cached_min_score is not None and ops_min_score <= cached_min_score: + return signals + return _filter_composite_min_score(signals, ops_min_score) + + def build_technique_params(settings: Settings) -> TechniqueParams: """운영용 기법 파라미터를 구성한다.""" extra: dict[str, Any] = {} @@ -80,21 +179,77 @@ def generate_raw_signals( last_price = float(df["close"].iloc[-1]) cache_path = settings.techniques_dir / f"{settings.ops_technique_id}.json" + latest_bar = len(df) - 1 + from_cache = False + refreshed_tail = False + if use_cache and cache_path.exists(): - cached = load_technique_result(cache_path) + cached = _load_technique_cached(cache_path) + raw_signals = list(cached.signals) + max_cached_bar = _max_signal_bar_index(raw_signals) + if latest_bar > max_cached_bar: + tail_bars = max(settings.ops_signal_tail_bars, 200) + offset = max(0, len(df) - tail_bars) + tail_df = df.iloc[offset:].copy() + technique = get_technique(settings.ops_technique_id) + params = build_technique_params(settings) + tail_result = run_technique(technique, tail_df, params, gt_result=None) + raw_signals = _merge_tail_signals( + raw_signals, + tail_result.signals, + offset, + max_cached_bar, + ) + refreshed_tail = True + updated = TechniqueResult( + technique_id=cached.technique_id, + technique_name=cached.technique_name, + category=cached.category, + causal=cached.causal, + description=cached.description, + params=cached.params, + signals=raw_signals, + legs=cached.legs, + summary=cached.summary, + pnl=cached.pnl, + alignment=cached.alignment, + ) + _store_technique_memory(cache_path, updated) + if settings.ops_persist_signal_cache: + save_technique_result(updated, settings.techniques_dir) + logger.info( + "신호 tail 갱신 %s: bar %s→%s signals %s→%s", + settings.ops_technique_id, + max_cached_bar, + latest_bar, + len(cached.signals), + len(raw_signals), + ) + raw_signals = _apply_ops_min_score( + cached.technique_id, + raw_signals, + float(cached.params.get("min_score")) if cached.params.get("min_score") is not None else None, + settings.ops_min_score, + ) + params = dict(cached.params) + if settings.ops_min_score is not None: + params["min_score"] = settings.ops_min_score + from_cache = not refreshed_tail return { "technique_id": cached.technique_id, "technique_name": cached.technique_name, - "params": cached.params, - "raw_signals": cached.signals, + "params": params, + "raw_signals": raw_signals, "data_end": data_end, "last_price": last_price, - "from_cache": True, + "from_cache": from_cache, + "signal_refresh": "tail" if refreshed_tail else "cache", + "latest_bar_index": latest_bar, } technique = get_technique(settings.ops_technique_id) - params = build_technique_params(settings) - result = run_technique(technique, df, params, gt_result=None) + params_obj = build_technique_params(settings) + result = run_technique(technique, df, params_obj, gt_result=None) return { "technique_id": result.technique_id, "technique_name": result.technique_name, @@ -103,6 +258,8 @@ def generate_raw_signals( "data_end": data_end, "last_price": last_price, "from_cache": False, + "signal_refresh": "full", + "latest_bar_index": latest_bar, }