0ecac7cb95
한투 호가 = 2번째 앱키 전용 키 없거나 start 실패 시 메인에 H0STASP0 안 붙임. 운영설정 WS_ORDERBOOK_SAVE_KIS 빨간 danger. LS RAM 합집합 후보∪보유∪영구∪grace. sync_targets와 split reconcile 둘 다. 틱 DB 영구 게이트는 그대로. 분봉 쓰레기 → 다음 소스 봉 통째 그 분 틱 0건이거나 전부 봉끝 대비 LIVE_FEED_FALLBACK_MAX_AGE_SEC 초과면 구멍. 메인 WS → 2차 → LS → REST → rollup. CANDLE_GARBAGE_FALLBACK 기본 true. 파일: feed_fallback.py(신규), ws_manager.py, kis_ws.py, candle_series.py, bt_candle_source.py, live_config_schema.py, database.py, 스모크, MD 2개. 같은 ws_manager/database/kis_ws/live_config에는 직전 커밋 이후 쌓여 있던 시세 폴백·ENV 키 정리도 같이 들어갔습니다. 파일 단위로 나눌 수 없어서입니다.
87 lines
3.1 KiB
Python
Executable File
87 lines
3.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
scripts/apply_optuna_ob_consensus.py
|
|
====================================
|
|
Optuna 호가 후처리 합의(8방 중 best 또는 구JSON 축분리)를 DB env에 반영.
|
|
실행 시 recommend_orderbook_parameters()를 **새로 돌림** — JSON postprocess_consensus 와 별개.
|
|
|
|
실행 예시:
|
|
.venv/bin/python3 scripts/apply_optuna_ob_consensus.py --strategy BREAKOUT
|
|
.venv/bin/python3 scripts/apply_optuna_ob_consensus.py --strategy SCALP
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from database import TradeDB
|
|
from kis_trader.backtest.optuna_orderbook_recommend import (
|
|
build_orderbook_env_patch,
|
|
recommend_orderbook_parameters,
|
|
)
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description="Optuna 호가 합의 파라미터 DB 반영 CLI")
|
|
ap.add_argument("--strategy", default="MOMENTUM", help="반영 대상 전략명 (MOMENTUM, BREAKOUT, SCALP, TAIL)")
|
|
ap.add_argument("--n-trials", type=int, default=500, help="탐색 횟수 (기본 500회)")
|
|
args = ap.parse_args()
|
|
|
|
strat = args.strategy.upper()
|
|
print("=" * 75)
|
|
print(f"🔄 [Optuna 호가 합의 파라미터 DB 자동 반영 엔진]")
|
|
print(f" - 대상 전략 : {strat}")
|
|
print(f" - 탐색 규모 : {args.n_trials}회 고속 합의 탐색")
|
|
print("=" * 75)
|
|
|
|
rec = recommend_orderbook_parameters(strategy=strat, n_trials=args.n_trials)
|
|
if not rec.get("ok"):
|
|
print(f"❌ 합의 수치 도출 실패: 사유 = {rec.get('reason')}")
|
|
return 1
|
|
|
|
patch = build_orderbook_env_patch(rec)
|
|
if not patch:
|
|
print("❌ DB 패치 변환 실패: 대상 파라미터가 없습니다.")
|
|
return 1
|
|
|
|
db = TradeDB()
|
|
print("📥 [DB 반영 전 대상 파라미터 확인]")
|
|
for k, v in patch.items():
|
|
print(f" • {k} ➔ {v}")
|
|
print()
|
|
|
|
# DB 반영 (insert_env_snapshot)
|
|
db.insert_env_snapshot(patch)
|
|
print("✅ DB 환경변수 스키마에 패치 전송 완료.")
|
|
|
|
# 캐시 무효화 (실전/백테 즉시 반영)
|
|
try:
|
|
from kis_trader.utils.env import invalidate_merged_env_cache
|
|
invalidate_merged_env_cache()
|
|
except Exception as exc:
|
|
print(f"⚠️ 캐시 무효화 경고: {exc}")
|
|
|
|
# [검증 의무] DB 스냅샷 재조회 후 저장값 100% 정합성 검증
|
|
print("🛡️ [필수 검증] DB 최신 스냅샷 재조회 및 100% 정합성 검증 중...")
|
|
saved_env = db.get_merged_env_snapshot()
|
|
for k, expected_v in patch.items():
|
|
actual_v = str(saved_env.get(k, "")).lower()
|
|
exp_v = str(expected_v).lower()
|
|
if actual_v != exp_v:
|
|
print(f"❌ [검증 불일치] {k}: expected={exp_v}, actual={actual_v}")
|
|
return 1
|
|
print(f" ✅ {k} : {actual_v} (100% 일치)")
|
|
|
|
print("-" * 75)
|
|
print(f"👑 [{strat}] 호가 수급 필터 & 익절 가드 파라미터 DB 완공 및 검증 완결!")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|