dsyoon/Bithumb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: GT·시뮬·운영 3축 정리 및 hybrid 실거래 정합

Browse Source 
Phase C/dry-run·미사용 모듈·재생성 HTML을 제거하고, 운영 체결을
sim_causal_hybrid와 동일한 hybrid 로직으로 통합한다.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
xavis
2026-06-03 23:50:28 +09:00
parent a16c942be4
commit d7848df6f7
85 changed files with 177180 additions and 196131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
scripts/02_ground_truth.py
View File

@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""02단계: Ground Truth 매수·매도 타점 생성."""
"""Ground Truth: 매수·매도 정답 타점 JSON 생성."""
import runpy
from pathlib import Path

2
scripts/03_gt_mtf_profile.py
View File

@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""03c단계: GT 타점 MTF 프로필 분석 (3분~일봉, 매수/매도 대조)."""
"""03c단계: GT 타점 MTF 프로필 분석 (3분~월봉, 매수/매도 대조)."""
import runpy
from pathlib import Path

17
scripts/03_patch_gt_snapshots.py
View File

@@ -1,17 +0,0 @@
#!/usr/bin/env python3
"""03b CSV에 누락된 GT 타점만 MTF 스냅샷 보강."""
import runpy
from pathlib import Path
runpy.run_path(str(Path(__file__).resolve().parent / "_bootstrap.py"))
from config import CHART_LOOKBACK_DAYS, SYMBOL
from deepcoin.analysis.general_analysis_snapshot import append_missing_gt_snapshots
from deepcoin.data.mtf_bb import load_frames_from_db
from deepcoin.ops.monitor import Monitor
if __name__ == "__main__":
mon = Monitor(cooldown_file=None)
frames = load_frames_from_db(mon, SYMBOL, lookback_days=CHART_LOOKBACK_DAYS)
n = append_missing_gt_snapshots(frames)
print(f"완료: {n}건 추가")

11
scripts/04_calibrate_gt_assets.py
View File

@@ -1,11 +0,0 @@
#!/usr/bin/env python3
"""GT 총자산 90% 목표 MTF 프로필 반복 캘리브레이션 (03b 스냅샷)."""
import runpy
from pathlib import Path
runpy.run_path(str(Path(__file__).resolve().parent / "_bootstrap.py"))
from deepcoin.matching.gt_profile_iterate import run_profile_calibration_loop
if __name__ == "__main__":
run_profile_calibration_loop()

11
scripts/04_gt_comparison_report.py
View File

@@ -1,11 +0,0 @@
#!/usr/bin/env python3
"""GT(450타점) vs 규칙 발화·시뮬 Go/No-Go 비교 리포트."""
import runpy
from pathlib import Path
runpy.run_path(str(Path(__file__).resolve().parent / "_bootstrap.py"))
from deepcoin.matching.gt_comparison import run_gt_comparison_report
if __name__ == "__main__":
run_gt_comparison_report()

2
scripts/04_match_rules.py
View File

@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""04단계: GT 프로필 + 전구간 EV 매칭 (04a~04d)."""
"""Simulation: GT 프로필 + 인과 EV 매칭 (규칙·monitor_rules)."""
import runpy
from pathlib import Path

2
scripts/04_simulation_report.py
View File

@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""1단계: walk-forward·민감도·Go/No-Go 시뮬레이션 리포트."""
"""Simulation: walk-forward·Go/No-Go 시뮬 리포트."""
import runpy
from pathlib import Path

13
scripts/05_chart_bb.py
View File

@@ -1,13 +0,0 @@
#!/usr/bin/env python3
"""05: 3분봉 BB 차트 HTML."""
import runpy
import sys
from pathlib import Path
runpy.run_path(str(Path(__file__).resolve().parent / "_bootstrap.py"))
from deepcoin.ops import simulation
if __name__ == "__main__":
sys.argv = [sys.argv[0]]
simulation.main()

27
scripts/05_chart_truth.py
View File

@@ -1,13 +1,32 @@
#!/usr/bin/env python3
"""05: Ground Truth 차트 + JSON."""
"""05: Ground Truth JSON → HTML 차트 (기본: JSON만 반영, --regenerate 시 02 재실행)."""
import argparse
import runpy
import sys
from pathlib import Path
runpy.run_path(str(Path(__file__).resolve().parent / "_bootstrap.py"))
from deepcoin.ops import simulation
def main() -> None:
parser = argparse.ArgumentParser(description="Ground Truth HTML 차트")
parser.add_argument(
"--regenerate",
action="store_true",
help="DB에서 GT JSON을 다시 생성한 뒤 HTML 작성",
)
parser.add_argument(
"--no-browser",
action="store_true",
help="브라우저 자동 열기 생략",
)
args = parser.parse_args()
simulation.run_ground_truth_chart(
open_browser=not args.no_browser,
from_json=not args.regenerate,
)
if __name__ == "__main__":
sys.argv = [sys.argv[0], "truth"]
simulation.main()
main()

9
scripts/06_execute_live.py
View File

@@ -2,6 +2,7 @@
"""3단계: 실거래 (monitor_rules + 빗썸 주문). LIVE_TRADING_ENABLED=1 필수."""
import argparse
import runpy
import sys
from pathlib import Path
runpy.run_path(str(Path(__file__).resolve().parent / "_bootstrap.py"))
@@ -23,9 +24,8 @@ if __name__ == "__main__":
parser.add_argument("--once", action="store_true", help="1회만 실행")
args = parser.parse_args()
st = env_status()
mode = "LIVE=ON (실주문)" if LIVE_TRADING_ENABLED else "LIVE=OFF (dry-run)"
print(
f"[06] 운영 설정 · {mode} · "
f"[06] 운영 설정 · LIVE=ON · "
f"초기₩{GT_INITIAL_CASH_KRW:,} · 일한도₩{LIVE_DAILY_KRW_MAX:,} · "
f"일손실₩{LIVE_DAILY_LOSS_LIMIT_KRW:,} · max_trades={LIVE_MAX_TRADES_PER_DAY}"
)
@@ -40,9 +40,10 @@ if __name__ == "__main__":
)
if not LIVE_TRADING_ENABLED:
print(
"주의: LIVE_TRADING_ENABLED=0 — .env에 1인지 확인 후 재기동. "
"(ncue 등 dotenv 없는 환경이면 pip install python-dotenv)"
"오류: LIVE_TRADING_ENABLED=0 — dry-run은 제거되었습니다. "
".env 에 LIVE_TRADING_ENABLED=1 설정 후 재실행하세요."
)
sys.exit(1)
trader = LiveTrader()
if args.once:
trader.run_once()

200
scripts/06_verify_live_dryrun.py → scripts/06_verify_live.py
View File

@@ -1,8 +1,9 @@
#!/usr/bin/env python3
"""Phase A: live_trader dry-run·sim_causal_hybrid(06) 정합·한도 점검."""
"""실거래(06) 기동 전 설정·hybrid tier·규칙·한도 점검. LIVE_TRADING_ENABLED=1 필수."""
from __future__ import annotations
import argparse
import runpy
from datetime import datetime
from pathlib import Path
@@ -13,6 +14,7 @@ from config import ( # noqa: E402
CHART_LOOKBACK_DAYS,
GT_INITIAL_CASH_KRW,
GT_SIGNAL_CAUSAL,
LIVE_HYBRID_BOOTSTRAP_FIRES,
LIVE_COOLDOWN_MIN,
LIVE_DAILY_KRW_MAX,
LIVE_DAILY_LOSS_LIMIT_KRW,
@@ -34,12 +36,13 @@ from deepcoin.ground_truth.causal_gt_hybrid import _close_series_from_df, hybrid
from deepcoin.ground_truth.gt_model import leg_entry_weights
from deepcoin.matching.position_sizing import compute_buy_amount_krw
from deepcoin.matching.load_rules import load_monitor_rules
from deepcoin.matching.live_sizing import LivePositionState, live_sizing_enabled
from deepcoin.ops.hybrid_sim_execution import replay_paper_portfolio
from deepcoin.ops.hybrid_sim_execution import (
bootstrap_monitor_signals_from_outcomes,
build_live_signal_history,
plan_live_hit,
)
from deepcoin.ops.live_trader import LiveTrader
from deepcoin.ops.monitor import Monitor
from deepcoin.ops.paper_portfolio import PaperPortfolio
from deepcoin.matching.position_sizing import load_ev_wf_approved_rule_ids
def _plan_with_dd(
@@ -84,7 +87,8 @@ def check_config() -> list[str]:
issues: list[str] = []
_print_header("1. 설정")
print(f" SYMBOL={SYMBOL}")
print(f" GT_SIGNAL_CAUSAL={GT_SIGNAL_CAUSAL} (live_sizing={live_sizing_enabled()})")
print(f" GT_SIGNAL_CAUSAL={GT_SIGNAL_CAUSAL}")
print(f" LIVE_HYBRID_BOOTSTRAP_FIRES={LIVE_HYBRID_BOOTSTRAP_FIRES}")
print(f" LIVE_TRADING_ENABLED={LIVE_TRADING_ENABLED}")
print(f" SIM_PRIMARY_SIZING={SIM_PRIMARY_SIZING}")
dd = load_hybrid_dd_params()
@@ -103,10 +107,12 @@ def check_config() -> list[str]:
print(f" GT_INITIAL_CASH_KRW=₩{GT_INITIAL_CASH_KRW:,}")
rules = load_monitor_rules()
print(f" monitor_rules={[r['rule_id'] for r in rules]}")
if not GT_SIGNAL_CAUSAL:
issues.append("GT_SIGNAL_CAUSAL=0 — hybrid live sizing 비활성")
if not LIVE_TRADING_ENABLED:
issues.append("LIVE_TRADING_ENABLED=0 — 실전 운영 시 1 필요")
issues.append("LIVE_TRADING_ENABLED=0 — 실거래 기동 불가")
if not GT_SIGNAL_CAUSAL:
issues.append("GT_SIGNAL_CAUSAL=0 — 인과 GT·hybrid 정합 권장(1)")
if not LIVE_HYBRID_BOOTSTRAP_FIRES:
issues.append("LIVE_HYBRID_BOOTSTRAP_FIRES=0 — 시뮬 이력 부트스트랩 꺼짐")
if len(rules) != 2:
issues.append(f"monitor_rules {len(rules)}개 (기대 2)")
expected = {"buy_compound_tight", "sell_mtf_cross_all_tf"}
@@ -117,31 +123,18 @@ def check_config() -> list[str]:
def check_capital_alignment() -> list[str]:
"""
초기 자금 40만 원 기준 원화 한도·알림 비율 점검.
LIVE_TRADING_ENABLED=1 → Phase B-1(운영), 0 → Phase C(dry-run).
Returns:
불일치 시 이슈 문자열 목록.
"""
"""초기 자금 40만 원 기준 원화 한도·알림 비율 점검."""
issues: list[str] = []
phase = "B-1 실전" if LIVE_TRADING_ENABLED else "C dry-run"
_print_header(f"1b. 초기 자금·비율 (40만 원, {phase})")
_print_header("1b. 초기 자금·비율 (운영)")
ic = int(GT_INITIAL_CASH_KRW)
expected: dict[str, int] = {
"GT_INITIAL_CASH_KRW": 400_000,
"MONITOR_ALERT_KRW_AMOUNT": int(ic * 0.10),
"LIVE_ORDER_KRW": int(ic * 0.10),
"LIVE_DAILY_LOSS_LIMIT_KRW": int(ic * 0.10),
"LIVE_DAILY_KRW_MAX": ic,
"LIVE_MAX_TRADES_PER_DAY": 15,
}
if LIVE_TRADING_ENABLED:
expected["LIVE_DAILY_LOSS_LIMIT_KRW"] = int(ic * 0.10)
expected["LIVE_DAILY_KRW_MAX"] = ic
expected["LIVE_MAX_TRADES_PER_DAY"] = 15
else:
expected["LIVE_DAILY_LOSS_LIMIT_KRW"] = int(ic * 0.05)
expected["LIVE_DAILY_KRW_MAX"] = int(ic * 10)
expected["LIVE_MAX_TRADES_PER_DAY"] = 999
actual = {
"GT_INITIAL_CASH_KRW": ic,
"MONITOR_ALERT_KRW_AMOUNT": int(MONITOR_ALERT_KRW_AMOUNT),
@@ -157,16 +150,6 @@ def check_capital_alignment() -> list[str]:
print(f" [{mark}] {key}={got:,} (기대 {exp:,})")
if not ok:
issues.append(f"{key}={got:,} ≠ 기대 {exp:,}")
if not LIVE_TRADING_ENABLED:
paper = PaperPortfolio.load()
if int(paper.cash_krw) != ic and paper.qty < 1e-12:
issues.append(
f"paper 현금 ₩{paper.cash_krw:,.0f} ≠ 초기 ₩{ic:,} (보유 없을 때)"
)
elif int(getattr(paper, "initial_cash_krw", 0) or paper.cash_krw) != ic:
print(f" [INFO] paper 운용 중 (cash=₩{paper.cash_krw:,.0f})")
else:
print(" [INFO] LIVE=1 — paper_portfolio 검사 생략 (실계좌·live_signal_history 사용)")
return issues
@@ -179,24 +162,14 @@ def check_tier_sizing(df) -> list[str]:
scenarios = [
("신규·소형DD(1%)", ic, 0.0, 1.0, {}),
("신규·대형DD(6%)", ic, 0.0, 6.0, {}),
("복리·과거large leg", ic * 5, 2000.0, 3.0, {"completed_leg_ret": {1: 25.0}}),
("복리·과거leg+", ic * 2, ic / price * 0.5, 3.0, {0: 12.0}),
]
for label, cash, qty, dd, extra in scenarios:
for label, cash, qty, dd, completed in scenarios:
hybrid_amt = _plan_with_dd(
cash,
qty,
price,
dd,
enhanced=False,
completed_leg_ret=extra.get("completed_leg_ret"),
cash, qty, price, dd, enhanced=False, completed_leg_ret=completed
)
conv_amt = _plan_with_dd(
cash,
qty,
price,
dd,
enhanced=True,
completed_leg_ret=extra.get("completed_leg_ret"),
cash, qty, price, dd, enhanced=True, completed_leg_ret=completed
)
print(
f" [{label}] cash={cash:,} hybrid={hybrid_amt:,} "
@@ -207,36 +180,30 @@ def check_tier_sizing(df) -> list[str]:
return issues
def check_paper_replay(df) -> list[str]:
"""paper signal_history → 시뮬 replay 잔고 일치."""
def check_hybrid_bootstrap(df) -> list[str]:
"""fire_outcomes 부트스트랩 건수·plan_live_hit 동작."""
issues: list[str] = []
_print_header("3. paper 시뮬 replay")
paper = PaperPortfolio.load()
hist = paper.signal_history
print(f" signal_history={len(hist)} · saved cash=₩{paper.cash_krw:,.0f} qty={paper.qty:.4f}")
if not hist:
print(" (이력 없음 — 신규 dry-run)")
_print_header("2b. hybrid 부트스트랩 (sim_causal_hybrid)")
boot = bootstrap_monitor_signals_from_outcomes()
print(f" fire_outcomes monitor 발화: {len(boot)}건")
if LIVE_HYBRID_BOOTSTRAP_FIRES and len(boot) < 100:
issues.append(f"bootstrap 발화 {len(boot)}건 — 04_match_rules 재실행 필요")
if df is None or df.empty or not boot:
return issues
approved = load_ev_wf_approved_rule_ids()
replayed, _ = replay_paper_portfolio(hist, df, approved_buy_rules=approved)
cash_diff = abs(replayed.cash_krw - paper.cash_krw)
qty_diff = abs(replayed.qty - paper.qty)
print(
f" replay cash=₩{replayed.cash_krw:,.0f} qty={replayed.qty:.4f} "
f"(Δcash={cash_diff:,.0f} Δqty={qty_diff:.6f})"
)
if cash_diff > 1.0 or qty_diff > 1e-6:
issues.append(
"paper_portfolio.json 과 sim replay 불일치 — "
"signal_history 갱신 후 06 --once 1회 권장"
hist = build_live_signal_history([])
last_buy = [s for s in boot if s["side"] == "buy"][-1:]
if last_buy:
plan = plan_live_hit(hist, last_buy[0], df, approved_buy_rules=None)
print(
f" 최근 매수 샘플 {last_buy[0]['dt']}: "
f"ok={plan.ok} ₩{plan.amount_krw:,.0f} ({plan.message})"
)
return issues
def check_live_limits() -> None:
"""시뮬 대비 실거래 일한도 영향 안내."""
_print_header("4. 실거래 한도 vs hybrid tier")
st = LivePositionState()
"""실거래 일한도 vs hybrid tier (plan_live_hit)."""
_print_header("3. 실거래 한도 vs hybrid tier")
mon = Monitor(cooldown_file=None)
frames = load_frames_from_db(mon, SYMBOL, lookback_days=CHART_LOOKBACK_DAYS)
df = frames.get(MATCH_PRIMARY_INTERVAL)
@@ -245,15 +212,19 @@ def check_live_limits() -> None:
close_s = _close_series_from_df(df)
if not close_s.empty:
price = float(close_s.iloc[-1])
planned = st.plan_buy_amount_krw(
str(df.index[-1]) if df is not None and not df.empty else "2026-06-01 12:00:00",
price,
float(GT_INITIAL_CASH_KRW),
0.0,
hist = build_live_signal_history([])
dt = str(df.index[-1]) if df is not None and not df.empty else "2026-06-01 12:00:00"
planned = plan_live_hit(
hist,
{
"dt": dt,
"rule_id": "buy_compound_tight",
"side": "buy",
"close": price,
},
df,
enhanced=False,
fee_rate=TRADING_FEE_RATE,
)
approved_buy_rules=None,
).amount_krw
trader = LiveTrader()
ok, reason = trader._can_trade("buy_compound_tight", planned)
print(f" 현재가={price:,.0f} · hybrid planned={planned:,}")
@@ -261,13 +232,13 @@ def check_live_limits() -> None:
if planned > LIVE_DAILY_KRW_MAX:
print(
" WARN: hybrid 1회 매수액이 일한도 초과 가능 — "
"파일럿은 의도적 제한, 시뮬(+1121%)과 괴리 발생"
"시뮬 대비 실거래 체결액이 작아질 수 있음"
)
def check_live_eval() -> None:
"""현재 시점 규칙 발화."""
_print_header("5. 현재 발화 (live_eval)")
_print_header("4. 현재 발화 (live_eval)")
fired = evaluate_live_rules(force_refresh=True)
if not fired:
print(" 발화 없음 (정상 — 신호 대기)")
@@ -276,40 +247,27 @@ def check_live_eval() -> None:
print(f" {hit['side']} {hit['rule_id']} @ {hit['dt']} close={hit['close']}")
def run_dryrun_once() -> None:
"""06 1회 dry-run."""
_print_header("6. 06_execute_live --once (dry-run)")
def run_live_once() -> None:
"""06 1회 실거래 루프 (발화 시 주문)."""
_print_header("5. 06_execute_live --once")
LiveTrader().run_once()
def write_verification_report(issues: list[str], out_path: Path) -> None:
"""Phase A 결과를 docs/05_ops에 기록."""
"""운영 점검 결과를 docs/05_ops에 기록."""
out_path.parent.mkdir(parents=True, exist_ok=True)
status = "PASS" if not issues else "WARN"
lines = [
"# Live Phase A — dry-run 검증",
"# Live 운영 점검",
"",
f"- 일시: {datetime.now():%Y-%m-%d %H:%M:%S}",
f"- 결과: **{status}**",
"",
"## Plan (목적)",
"",
"- 06 dry-run/live 체결이 `hybrid_sim_execution`(sim_causal_hybrid)과 정합인지 확인",
"- conviction tier(`enhanced=True`) 미사용 확인",
"- 실거래 한도가 hybrid tier와 어떻게 상호작용하는지 기록",
"",
"## Do (실행)",
"",
"```bash",
"python scripts/06_verify_live_dryrun.py",
"python scripts/06_execute_live.py --once",
"```",
"",
"## Check (점검 결과)",
"## Check",
"",
f"- GT_SIGNAL_CAUSAL={GT_SIGNAL_CAUSAL}",
f"- LIVE_TRADING_ENABLED={LIVE_TRADING_ENABLED}",
f"- monitor_rules: buy_compound_tight, sell_mtf_cross_all_tf",
"- monitor_rules: buy_compound_tight, sell_mtf_cross_all_tf",
f"- hybrid DD: {load_hybrid_dd_params()}",
"",
]
@@ -321,15 +279,15 @@ def write_verification_report(issues: list[str], out_path: Path) -> None:
lines.append("")
lines.extend(
[
"## Act (다음 단계)",
"## Act",
"",
"1. `05_run_monitor.py` 1~2일 병행 (알림만)",
"2. `.env` 파일럿 한도 확정 후 `LIVE_TRADING_ENABLED=1`",
"3. 1~2주 실계좌 PnL·슬리피지 기록 (본 문서 갱신)",
"```bash",
"python scripts/06_execute_live.py",
"```",
"",
"## Kill switch",
"",
"- `LIVE_TRADING_ENABLED=0` + 06 프로세스 중지",
"- 06 프로세스 중지",
"- 빗썸 앱 수동 청산",
"",
]
@@ -339,8 +297,20 @@ def write_verification_report(issues: list[str], out_path: Path) -> None:
def main() -> int:
"""Phase A 검증 실행."""
print("[06_verify] Phase A dry-run 검증 시작")
"""실거래 기동 전 점검."""
parser = argparse.ArgumentParser(description="06 실거래 점검")
parser.add_argument(
"--once",
action="store_true",
help="점검 후 06_execute_live 1회 실행 (실주문 가능)",
)
args = parser.parse_args()
print("[06_verify] 실거래 점검 시작")
if not LIVE_TRADING_ENABLED:
print("오류: LIVE_TRADING_ENABLED=0 — .env 에 1 설정 필요")
return 1
issues = check_config()
issues.extend(check_capital_alignment())
mon = Monitor(cooldown_file=None)
@@ -350,10 +320,11 @@ def main() -> int:
issues.append("3m OHLC 없음 — 01_download 필요")
else:
issues.extend(check_tier_sizing(df))
issues.extend(check_paper_replay(df))
issues.extend(check_hybrid_bootstrap(df))
check_live_limits()
check_live_eval()
run_dryrun_once()
if args.once:
run_live_once()
out = (
Path(__file__).resolve().parents[1]
/ "docs"
@@ -365,9 +336,8 @@ def main() -> int:
if issues:
for i in issues:
print(f" WARN: {i}")
print(" → 이슈 확인 후 Phase B(소액 파일럿) 진행")
return 1
print(" 운영 설정 PASS — 06_execute_live 실전 기동 가능")
print(" 운영 설정 PASS — 06_execute_live 기동 가능")
return 0

348
scripts/07_phase_c_paper_report.py
View File

@@ -1,348 +0,0 @@
#!/usr/bin/env python3
"""
Phase C dry-run 종료 후 모의 수익률(참고) 집계.
- 입력: data/ops/paper_fires.jsonl (06 dry-run 발화 로그)
- 출력: docs/05_ops/phase_c_paper_report.json + 콘솔 요약
주의: 실계좌 수익이 아님. 발화가 N봉 후 가격으로 계산한 forward % 합산(참고).
hybrid 복리 PnL은 simulation_report.html 과 다릅니다.
"""
from __future__ import annotations
import argparse
import json
import runpy
from datetime import datetime
from pathlib import Path
import numpy as np
import pandas as pd
runpy.run_path(str(Path(__file__).resolve().parent / "_bootstrap.py"))
from config import ( # noqa: E402
MATCH_FORWARD_BARS,
MATCH_PRIMARY_INTERVAL,
SYMBOL,
TRADING_FEE_RATE,
)
from deepcoin.matching.label_outcomes import _forward_ret_vectorized # noqa: E402
from deepcoin.ops.monitor import Monitor # noqa: E402
from deepcoin.ops.paper_portfolio import PaperPortfolio # noqa: E402
from deepcoin.paths import ( # noqa: E402
PAPER_FIRES_LOG,
PAPER_WEEKLY_REPORT_JSON,
PHASE_C_DAILY_DIR,
)
_FEE_PCT = TRADING_FEE_RATE * 2 * 100
def load_paper_fires(path: Path) -> pd.DataFrame:
"""paper_fires.jsonl → DataFrame."""
if not path.is_file():
return pd.DataFrame()
rows = []
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if line:
rows.append(json.loads(line))
if not rows:
return pd.DataFrame()
return pd.DataFrame(rows)
def attach_forward_returns(fires: pd.DataFrame, close_df: pd.DataFrame) -> pd.DataFrame:
"""
would_trade=True 발화에 MATCH_FORWARD_BARS 기준 forward 수익률(%) 부여.
Args:
fires: paper 발화.
close_df: 3분 종가 (datetime index).
Returns:
forward_ret_pct 컬럼 추가.
"""
if fires.empty or close_df.empty:
fires["forward_ret_pct"] = np.nan
return fires
close_df = close_df.sort_index()
if not isinstance(close_df.index, pd.DatetimeIndex):
close_df.index = pd.to_datetime(close_df.index)
close_ts_ns = close_df.index.astype(np.int64).values
close_px = close_df["Close"].astype(float).values
sub = fires[fires["would_trade"] == True].copy() # noqa: E712
if sub.empty:
fires["forward_ret_pct"] = np.nan
return fires
sig = pd.to_datetime(sub["signal_dt"])
fire_ns = sig.astype(np.int64).values
c0 = sub["close"].astype(float).values
side = sub["side"].astype(str).values
ret, valid = _forward_ret_vectorized(
fire_ns, c0, close_ts_ns, close_px, side, MATCH_FORWARD_BARS, _FEE_PCT
)
fires = fires.copy()
fires["forward_ret_pct"] = np.nan
idx = sub.index
fires.loc[idx, "forward_ret_pct"] = np.where(valid, ret, np.nan)
return fires
def summarize(fires: pd.DataFrame, *, report_kind: str = "daily") -> dict:
"""
집계 dict.
Args:
fires: forward_ret_pct 포함 발화 DataFrame.
report_kind: daily | final.
Returns:
JSON 직렬화 가능 dict.
"""
traded = fires[fires["would_trade"] == True] # noqa: E712
with_ret = traded[traded["forward_ret_pct"].notna()]
out: dict = {
"generated_at": datetime.now().isoformat(timespec="seconds"),
"report_kind": report_kind,
"symbol": SYMBOL,
"forward_bars": MATCH_FORWARD_BARS,
"fee_round_trip_pct": _FEE_PCT,
"total_signals": int(len(fires)),
"would_trade_count": int(len(traded)),
"skipped_count": int(len(fires) - len(traded)),
"labeled_count": int(len(with_ret)),
"note": (
"forward %는 발화별 참고 지표. "
"총보유금액(equity)은 paper_portfolio 모의 체결 기준."
),
}
if not fires.empty and "ts" in fires.columns:
out["log_from"] = str(fires["ts"].min())
out["log_to"] = str(fires["ts"].max())
buy_n = int((traded["side"] == "buy").sum()) if not traded.empty else 0
sell_n = int((traded["side"] == "sell").sum()) if not traded.empty else 0
out["buy_fires"] = buy_n
out["sell_fires"] = sell_n
if not with_ret.empty:
out["mean_forward_ret_pct"] = round(float(with_ret["forward_ret_pct"].mean()), 4)
out["sum_forward_ret_pct"] = round(float(with_ret["forward_ret_pct"].sum()), 4)
by_side = (
with_ret.groupby("side")["forward_ret_pct"]
.agg(["count", "mean", "sum"])
.round(4)
)
out["by_side"] = {k: v.to_dict() for k, v in by_side.iterrows()}
by_rule = (
traded.groupby("rule_id")
.size()
.to_dict()
if not traded.empty
else {}
)
out["fires_by_rule"] = by_rule
return out
def build_phase_c_report(
fires_path: Path | None = None,
*,
report_kind: str = "daily",
) -> tuple[dict, pd.DataFrame]:
"""
paper_fires 로드 → forward % → 리포트 dict.
Args:
fires_path: jsonl 경로 (기본 PAPER_FIRES_LOG).
report_kind: daily | final.
Returns:
(report, fires_with_returns) — 발화 없으면 ({}, empty DataFrame).
"""
path = fires_path or PAPER_FIRES_LOG
fires = load_paper_fires(path)
if fires.empty:
return {}, fires
mon = Monitor(cooldown_file=None)
df = mon.read_candles_from_db(SYMBOL, MATCH_PRIMARY_INTERVAL, max_rows=50000)
if df.empty:
df = mon.get_coin_some_data(SYMBOL, MATCH_PRIMARY_INTERVAL)
if not isinstance(df.index, pd.DatetimeIndex):
df = df.set_index(pd.to_datetime(df["datetime"]))
fires = attach_forward_returns(fires, df)
report = summarize(fires, report_kind=report_kind)
mark = float(df["Close"].iloc[-1]) if not df.empty and "Close" in df.columns else 0.0
paper = PaperPortfolio.load()
report["paper_portfolio"] = paper.summary(mark)
return report, fires
def format_report_text(report: dict) -> str:
"""사람이 읽기 쉬운 요약 텍스트."""
kind = report.get("report_kind", "daily")
title = "Phase C 최종 보고" if kind == "final" else "Phase C 중간 보고"
lines = [
f"=== {title} ({report.get('generated_at', '')}) ===",
f"심볼: {report.get('symbol', '')}",
]
pf = report.get("paper_portfolio") or {}
if pf:
lines.extend(
[
"--- 모의 계좌 (dry-run, 빗썸 잔고 미사용) ---",
f"초기 자금: ₩{pf.get('initial_cash_krw', 0):,.0f}",
f"현금: ₩{pf.get('cash_krw', 0):,.0f} · "
f"보유 {pf.get('qty', 0):.4f} {report.get('symbol', '')} "
f"(평가단가 ₩{pf.get('mark_price', 0):,.0f})",
f"코인 평가: ₩{pf.get('coin_value_krw', 0):,.0f}",
f"총보유금액: ₩{pf.get('equity_krw', 0):,.0f} "
f"(손익 ₩{pf.get('pnl_krw', 0):+,.0f} / {pf.get('pnl_pct', 0):+.2f}%)",
]
)
lines.append(
f"발화 합계: {report.get('total_signals', 0)} "
f"(체결 {report.get('would_trade_count', 0)}, "
f"매수 {report.get('buy_fires', 0)} / 매도 {report.get('sell_fires', 0)})"
)
if "log_from" in report:
lines.append(f"로그 구간: {report['log_from']} ~ {report['log_to']}")
if "sum_forward_ret_pct" in report:
lines.append(
f"모의 forward 합산: {report['sum_forward_ret_pct']}% "
f"(평균 {report['mean_forward_ret_pct']}%, "
f"{report.get('forward_bars')}봉 후, 참고용)"
)
else:
lines.append("모의 forward: 라벨 가능 건 없음 (봉·발화 부족)")
lines.append(report.get("note", ""))
return "\n".join(lines)
def write_report_outputs(
report: dict,
*,
json_path: Path | None = None,
text_path: Path | None = None,
) -> None:
"""JSON·텍스트 리포트 저장."""
if json_path:
json_path.parent.mkdir(parents=True, exist_ok=True)
json_path.write_text(
json.dumps(report, ensure_ascii=False, indent=2),
encoding="utf-8",
)
if text_path:
text_path.parent.mkdir(parents=True, exist_ok=True)
text_path.write_text(format_report_text(report), encoding="utf-8")
def append_verification_log(report: dict, verification_md: Path) -> None:
"""live_verification 일별 표 해당 날짜 행 갱신."""
if not verification_md.is_file():
return
text = verification_md.read_text(encoding="utf-8")
iso = report.get("generated_at", "")[:10]
try:
y, m, d = iso.split("-")
day_label = f"{int(m)}/{int(d)}"
except ValueError:
return
buy = report.get("buy_fires", 0)
sell = report.get("sell_fires", 0)
pf = report.get("paper_portfolio") or {}
equity = pf.get("equity_krw", "-")
pnl_pct = pf.get("pnl_pct", "-")
kind = report.get("report_kind", "daily")
memo = "C 최종" if kind == "final" else "중간보고"
row = (
f"| {day_label} | Y | - | Y | {buy} | {sell} | "
f"총₩{equity} ({pnl_pct}%) {memo} |"
)
marker = "### 일별 기록"
if marker not in text:
return
head, table = text.split(marker, 1)
lines = table.splitlines()
new_lines: list[str] = []
replaced = False
for line in lines:
if line.startswith(f"| {day_label} |"):
new_lines.append(row)
replaced = True
else:
new_lines.append(line)
if not replaced:
new_lines.append(row)
verification_md.write_text(head + marker + "\n".join(new_lines), encoding="utf-8")
def run_report(
*,
report_kind: str = "daily",
stamp: str | None = None,
update_verification: bool = True,
) -> dict:
"""
리포트 생성·저장·콘솔 출력.
Args:
report_kind: daily | final.
stamp: 파일명용 타임스탬프 (기본 now).
update_verification: live_verification md 갱신 여부.
Returns:
report dict (빈 dict 가능).
"""
report, fires = build_phase_c_report(report_kind=report_kind)
if not report:
print(f"[07] 발화 로그 없음: {PAPER_FIRES_LOG}")
print(" Phase C 기간 06_execute_live.py (LIVE=0) 상시 실행 후 재시도")
return {}
stamp = stamp or datetime.now().strftime("%Y%m%d_%H%M")
daily_dir = PHASE_C_DAILY_DIR
write_report_outputs(
report,
json_path=daily_dir / f"report_{stamp}_{report_kind}.json",
text_path=daily_dir / f"report_{stamp}_{report_kind}.txt",
)
write_report_outputs(report, json_path=PAPER_WEEKLY_REPORT_JSON)
if update_verification:
append_verification_log(
report,
Path(__file__).resolve().parents[1]
/ "docs/05_ops/live_verification_20260601.md",
)
print(format_report_text(report))
print(f"[07] JSON: {PAPER_WEEKLY_REPORT_JSON}")
print(f"[07] 일별: {daily_dir}/report_{stamp}_{report_kind}.*")
return report
def main() -> None:
"""CLI: paper_fires → forward % → 리포트 저장."""
parser = argparse.ArgumentParser(description="Phase C 모의 forward % 집계")
parser.add_argument(
"--kind",
choices=("daily", "final"),
default="daily",
help="daily=중간, final=금요일 최종",
)
parser.add_argument("--no-verification-md", action="store_true")
args = parser.parse_args()
run_report(
report_kind=args.kind,
update_verification=not args.no_verification_md,
)
if __name__ == "__main__":
main()

173
scripts/08_phase_c_supervisor.py
View File

@@ -1,173 +0,0 @@
#!/usr/bin/env python3
"""
Phase C 슈퍼바이저: 06 dry-run 상시 + 매일 22:00 중간보고 + 금요일 22:00 최종 후 종료.
사용:
python scripts/08_phase_c_supervisor.py
python scripts/08_phase_c_supervisor.py --end-date 2026-06-05 --report-hour 22
"""
from __future__ import annotations
import argparse
import runpy
import signal
import subprocess
import sys
import time
from datetime import date, datetime, timedelta
from pathlib import Path
_ROOT = Path(__file__).resolve().parents[1]
runpy.run_path(str(_ROOT / "scripts" / "_bootstrap.py"))
from config import LIVE_TRADING_ENABLED # noqa: E402
from deepcoin.paths import ( # noqa: E402
PHASE_C_SUPERVISOR_LOG,
PHASE_C_SUPERVISOR_PID,
)
_DEFAULT_PY = "/Users/dsyoon/opt/anaconda3/envs/coin/bin/python"
_REPORT_WINDOW_MIN = 5
def _log(msg: str) -> None:
"""슈퍼바이저 로그 (파일; nohup 시 stdout 중복 방지)."""
line = f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}"
PHASE_C_SUPERVISOR_LOG.parent.mkdir(parents=True, exist_ok=True)
with PHASE_C_SUPERVISOR_LOG.open("a", encoding="utf-8") as f:
f.write(line + "\n")
def _run_script(py: str, name: str, *args: str) -> int:
"""scripts/ 하위 스크립트 실행."""
cmd = [py, str(_ROOT / "scripts" / name), *args]
_log(f"실행: {' '.join(cmd)}")
proc = subprocess.run(cmd, cwd=str(_ROOT), check=False)
return proc.returncode
def _in_report_window(now: datetime, hour: int) -> bool:
"""보고 시각(시) 직후 REPORT_WINDOW_MIN 분 이내."""
return now.hour == hour and now.minute < _REPORT_WINDOW_MIN
def _stop_child(proc: subprocess.Popen[bytes] | None) -> None:
"""06 자식 프로세스 종료."""
if proc is None or proc.poll() is not None:
return
_log(f"06 종료 요청 pid={proc.pid}")
proc.send_signal(signal.SIGTERM)
try:
proc.wait(timeout=15)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait(timeout=5)
_log("06 종료 완료")
def _write_pid() -> None:
"""슈퍼바이저 PID 기록."""
PHASE_C_SUPERVISOR_PID.parent.mkdir(parents=True, exist_ok=True)
PHASE_C_SUPERVISOR_PID.write_text(str(os_getpid()), encoding="utf-8")
def os_getpid() -> int:
"""현재 PID."""
import os
return os.getpid()
def _daily_pipeline(py: str, *, final: bool) -> None:
"""다운로드 → verify → 07 보고."""
_run_script(py, "01_download.py")
_run_script(py, "06_verify_live_dryrun.py")
kind = "final" if final else "daily"
_run_script(py, "07_phase_c_paper_report.py", "--kind", kind)
def main() -> int:
"""
Phase C 슈퍼바이저 메인.
Returns:
종료 코드 0=정상, 1=설정 오류.
"""
parser = argparse.ArgumentParser(description="Phase C dry-run 슈퍼바이저")
parser.add_argument(
"--end-date",
type=lambda s: date.fromisoformat(s),
default=date(2026, 6, 5),
help="최종 보고·종료일 (금요일, ISO)",
)
parser.add_argument("--report-hour", type=int, default=22, help="일일 보고 시각(시, 24h)")
parser.add_argument(
"--py",
default=_DEFAULT_PY,
help="Python 실행 파일 (coin conda)",
)
args = parser.parse_args()
if LIVE_TRADING_ENABLED:
_log("오류: LIVE_TRADING_ENABLED=1 — Phase C는 0 이어야 합니다.")
return 1
_write_pid()
reported: set[date] = set()
py = args.py
end: date = args.end_date
hour: int = args.report_hour
_log(
f"Phase C 슈퍼바이저 시작 · end={end} · 보고 {hour}:00 KST · LIVE=0"
)
child = subprocess.Popen(
[py, str(_ROOT / "scripts" / "06_execute_live.py")],
cwd=str(_ROOT),
)
_log(f"06 dry-run 기동 pid={child.pid}")
try:
while True:
now = datetime.now()
today = now.date()
if child.poll() is not None:
_log(f"06 비정상 종료 code={child.returncode} — 재기동")
child = subprocess.Popen(
[py, str(_ROOT / "scripts" / "06_execute_live.py")],
cwd=str(_ROOT),
)
if _in_report_window(now, hour) and today not in reported:
if today <= end:
reported.add(today)
is_final = today == end
_log(
f"{'최종' if is_final else '중간'} 보고 시작 ({today})"
)
_daily_pipeline(py, final=is_final)
if is_final:
_log("금요일 최종 보고 완료 — Phase C dry-run 종료")
break
if today > end and today not in reported:
_log("종료일 경과 — 최종 보고(미실시 시) 후 종료")
_daily_pipeline(py, final=True)
break
time.sleep(60)
except KeyboardInterrupt:
_log("KeyboardInterrupt — 종료")
finally:
_stop_child(child)
if PHASE_C_SUPERVISOR_PID.is_file():
PHASE_C_SUPERVISOR_PID.unlink(missing_ok=True)
_log("슈퍼바이저 종료")
return 0
if __name__ == "__main__":
sys.exit(main())

62
scripts/README.md
View File

@@ -1,20 +1,54 @@
# scripts — 단계별 CLI
# scripts — CLI (Ground Truth · Simulation · Operations)
프로젝트 루트에서 실행하세요. 각 스크립트는 `_bootstrap.py`로 `.env`와 `config`를 로드합니다.
`_bootstrap.py`가 `.env`를 로드합니다. 설계: [docs/reference/ARCHITECTURE.md](../docs/reference/ARCHITECTURE.md)
## 데이터 (공통)
| 스크립트 | 설명 |
|----------|------|
| `01_download.py` | OHLCV 적재 |
| `00_sync_ops.py` | 운영 전 DB 동기화 |
## Ground Truth
| 스크립트 | 설명 |
|----------|------|
| `02_ground_truth.py` | 정답 타점 JSON |
| `03_analyze_enrich.py` | 10TF 지표 enrich |
| `03_analyze_trades.py` | GT 타점 MTF 스냅샷 CSV |
| `03_gt_mtf_profile.py` | GT MTF 프로필 |
| `05_chart_truth.py` | GT HTML 차트 |
## Simulation
| 스크립트 | 설명 |
|----------|------|
| `04_match_rules.py` | 규칙·`matched_rules.json` |
| `04_simulation_report.py` | Go/No-Go 리포트 |
| `04_hybrid_dd_calibrate.py` | hybrid DD tier (필요 시) |
| `04_causal_gt_calibrate.py` | 인과 GT leg (필요 시) |
## Operations
| 스크립트 | 설명 |
|----------|------|
| `06_verify_live.py` | 기동 전 점검 |
| `check_balance.py` | 빗썸 잔고 |
| `06_execute_live.py` | 실거래 |
| `05_run_monitor.py` | 알림만 (선택) |
## 검증
| 스크립트 | 설명 |
|----------|------|
| `verify_env.py` | `.env` 검증 |
| `test_buy_sell_rehearsal.py` | hybrid 체결 제약 |
## 재구축
```bash
python scripts/01_download.py
python scripts/02_ground_truth.py
python scripts/03_analyze_enrich.py
python scripts/03_analyze_trades.py
python scripts/04_match_rules.py
python scripts/04_simulation_report.py # 1단계 Go/No-Go
python scripts/05_run_monitor.py # 알림만
python scripts/06_verify_live_dryrun.py # Phase A dry-run·tier 점검
python scripts/06_execute_live.py # 3단계 실거래
python scripts/05_chart_truth.py
python scripts/05_chart_bb.py
python scripts/verify_env.py
python scripts/03_analyze_enrich.py && python scripts/03_analyze_trades.py && python scripts/03_gt_mtf_profile.py
python scripts/04_match_rules.py && python scripts/04_simulation_report.py
```
상세 구조: [docs/reference/STRUCTURE.md](../docs/reference/STRUCTURE.md)

20
scripts/run.ps1
View File

@@ -1,20 +0,0 @@
# DeepCoin CLI wrapper — conda 환경 ncue 고정
# 사용: .\scripts\run.ps1 06_execute_live.py --once
param(
[Parameter(Mandatory = $true, Position = 0)]
[string]$Script,
[Parameter(ValueFromRemainingArguments = $true)]
[string[]]$Rest
)
$ErrorActionPreference = "Stop"
$root = Split-Path $PSScriptRoot -Parent
Set-Location $root
$env:PYTHONUNBUFFERED = "1"
$scriptPath = Join-Path $root "scripts" $Script
if (-not (Test-Path $scriptPath)) {
throw "스크립트 없음: $scriptPath"
}
& conda run -n ncue --no-capture-output python $scriptPath @Rest

17
scripts/run_phase_c.sh
View File

@@ -1,17 +0,0 @@
#!/bin/bash
# Phase C dry-run (~금요일 저녁). 프로젝트 루트에서 실행.
set -e
cd "$(dirname "$0")/.."
PY="${PY:-/Users/dsyoon/opt/anaconda3/envs/coin/bin/python}"
echo "=== Phase C: LIVE_TRADING_ENABLED=0 확인 ==="
$PY -c "import runpy; from pathlib import Path; runpy.run_path('scripts/_bootstrap.py'); import config; assert not config.LIVE_TRADING_ENABLED, 'LIVE must be 0'"
echo "=== 01 봉 갱신 (1일 1회 권장) ==="
$PY scripts/01_download.py
echo "=== verify ==="
$PY scripts/06_verify_live_dryrun.py
echo "=== 06 dry-run 상시 (Ctrl+C 종료) ==="
exec $PY scripts/06_execute_live.py

22
scripts/run_phase_c_supervised.sh
View File

@@ -1,22 +0,0 @@
#!/bin/bash
# Phase C: 슈퍼바이저(06 상시 + 22시 보고 + 금요일 종료). 백그라운드 기동용.
set -e
cd "$(dirname "$0")/.."
PY="${PY:-/Users/dsyoon/opt/anaconda3/envs/coin/bin/python}"
LOG="${LOG:-data/ops/phase_c_supervisor.log}"
PIDFILE="${PIDFILE:-data/ops/phase_c_supervisor.pid}"
if [[ -f "$PIDFILE" ]]; then
old=$(cat "$PIDFILE")
if kill -0 "$old" 2>/dev/null; then
echo "이미 실행 중 (pid=$old). 중복 기동하지 않습니다."
exit 0
fi
fi
mkdir -p data/ops
nohup "$PY" -u scripts/08_phase_c_supervisor.py >> "$LOG" 2>&1 &
disown
echo $! > "$PIDFILE"
echo "Phase C 슈퍼바이저 기동 pid=$(cat "$PIDFILE")"
echo "로그: $LOG"

47
scripts/test_buy_sell_rehearsal.py
View File

@@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
40만 원 기준 매수·매도 최종 리허설 (DB 없이 synthetic + paper replay).
40만 원 기준 매수·매도 리허설 (DB 없이 synthetic + hybrid replay).
"""
from __future__ import annotations
@@ -13,14 +13,13 @@ runpy.run_path(str(Path(__file__).resolve().parent / "_bootstrap.py"))
import pandas as pd
from config import GT_INITIAL_CASH_KRW, TRADING_FEE_RATE
from deepcoin.matching.position_sizing import load_ev_wf_approved_rule_ids
from deepcoin.ops.hybrid_sim_execution import (
HybridSimPortfolio,
hit_key,
plan_live_hit,
replay_paper_portfolio,
replay_hybrid_signals,
sort_hits_sim_order,
)
from deepcoin.ops.paper_portfolio import PaperPortfolio
def _mini_ohlc() -> pd.DataFrame:
@@ -47,7 +46,7 @@ def test_sort_buy_before_sell() -> None:
def test_sell_without_holdings() -> None:
"""보유 없이 매도만 → 모의 보유 없음."""
ohlc = _mini_ohlc()
approved = load_ev_wf_approved_rule_ids()
approved = None
hist = [
{
"dt": "2026-06-01 12:00:00",
@@ -56,18 +55,18 @@ def test_sell_without_holdings() -> None:
"close": 500.0,
}
]
paper, results = replay_paper_portfolio(hist, ohlc, approved_buy_rules=approved)
portfolio, results = replay_hybrid_signals(hist, ohlc, approved_buy_rules=approved)
key = hit_key(hist[0])
res = results[key]
assert not res.ok and "보유 없음" in res.message, res
assert paper.qty < 1e-9 and paper.cash_krw == float(GT_INITIAL_CASH_KRW)
assert portfolio.qty < 1e-9 and portfolio.cash_krw == float(GT_INITIAL_CASH_KRW)
print(" [OK] 보유 없음 매도 스킵")
def test_buy_then_partial_sell() -> None:
"""매수 후 분할 매도 1회."""
ohlc = _mini_ohlc()
approved = load_ev_wf_approved_rule_ids()
approved = None
dt_buy = str(ohlc.index[50])
dt_sell = str(ohlc.index[80])
price_buy = float(ohlc.loc[ohlc.index[50], "Close"])
@@ -86,22 +85,22 @@ def test_buy_then_partial_sell() -> None:
"close": price_sell,
},
]
paper, results = replay_paper_portfolio(hist, ohlc, approved_buy_rules=approved)
portfolio, results = replay_hybrid_signals(hist, ohlc, approved_buy_rules=approved)
buy_res = results[hit_key(hist[0])]
sell_res = results[hit_key(hist[1])]
assert buy_res.ok, buy_res.message
assert paper.qty > 0 or sell_res.ok, (paper.qty, sell_res)
assert portfolio.qty > 0 or sell_res.ok, (portfolio.qty, sell_res)
if sell_res.ok:
assert sell_res.sell_qty > 0 and sell_res.amount_krw > 0
assert paper.cash_krw > float(GT_INITIAL_CASH_KRW) * 0.5
assert portfolio.cash_krw > float(GT_INITIAL_CASH_KRW) * 0.5
print(
f" [OK] 매수 ₩{buy_res.amount_krw:,.0f} → 매도 "
f"ok={sell_res.ok} qty={sell_res.sell_qty:.4f} 현금=₩{paper.cash_krw:,.0f}"
f"ok={sell_res.ok} qty={sell_res.sell_qty:.4f} 현금=₩{portfolio.cash_krw:,.0f}"
)
def test_unapproved_buy_excluded_from_sizing() -> None:
"""EV/WF 미포함 매수는 hybrid 배분 입력에서 제외."""
def test_unapproved_buy_excluded_when_filter_set() -> None:
"""approved_buy_rules 지정 시에만 미포함 매수 제외 (시뮬 기본은 필터 없음)."""
ohlc = _mini_ohlc()
hist = [
{
@@ -112,15 +111,15 @@ def test_unapproved_buy_excluded_from_sizing() -> None:
},
]
approved = {"buy_compound_tight"}
sized_hist = replay_paper_portfolio(hist, ohlc, approved_buy_rules=approved)[0]
sized_hist = replay_hybrid_signals(hist, ohlc, approved_buy_rules=approved)[0]
assert sized_hist.qty < 1e-9
print(" [OK] 미승인 매수 규칙 → 체결 없음")
print(" [OK] approved_buy_rules 지정 시 미승인 매수 제외")
def test_plan_live_matches_replay() -> None:
"""plan_live_hit == replay 마지막 건."""
ohlc = _mini_ohlc()
approved = load_ev_wf_approved_rule_ids()
approved = None
hist = []
hit = {
"dt": str(ohlc.index[70]),
@@ -130,7 +129,7 @@ def test_plan_live_matches_replay() -> None:
}
plan = plan_live_hit(hist, hit, ohlc, approved_buy_rules=approved)
hist.append(hit)
_, results = replay_paper_portfolio(hist, ohlc, approved_buy_rules=approved)
_, results = replay_hybrid_signals(hist, ohlc, approved_buy_rules=approved)
replay_res = results[hit_key(hit)]
assert plan.amount_krw == replay_res.amount_krw, (plan, replay_res)
assert plan.ok == replay_res.ok
@@ -140,7 +139,7 @@ def test_plan_live_matches_replay() -> None:
def test_initial_cash_400k_large_buy() -> None:
"""40만·대형 DD 시 매수액 ≤ 가용현금."""
ohlc = _mini_ohlc()
approved = load_ev_wf_approved_rule_ids()
approved = None
hit = {
"dt": str(ohlc.index[100]),
"rule_id": "buy_compound_tight",
@@ -155,9 +154,9 @@ def test_initial_cash_400k_large_buy() -> None:
print(f" [OK] 40만 대형 tier 매수 ₩{plan.amount_krw:,.0f} (≤{GT_INITIAL_CASH_KRW:,})")
def test_paper_apply_buy_insufficient() -> None:
def test_apply_buy_insufficient_cash() -> None:
"""현금 부족 시 apply_buy 실패."""
p = PaperPortfolio()
p = HybridSimPortfolio()
p.cash_krw = 10_000.0
ok = p.apply_buy(50_000, 500.0, leg_id=1)
assert not ok
@@ -167,16 +166,16 @@ def test_paper_apply_buy_insufficient() -> None:
def main() -> int:
"""리허설 실행."""
print(f"[리허설] GT_INITIAL_CASH_KRW=₩{GT_INITIAL_CASH_KRW:,}")
print(f" approved buys: {load_ev_wf_approved_rule_ids()}")
print(" approved_buy_rules: None (sim_causal_hybrid 동일)")
fails = 0
tests = [
test_sort_buy_before_sell,
test_sell_without_holdings,
test_buy_then_partial_sell,
test_unapproved_buy_excluded_from_sizing,
test_unapproved_buy_excluded_when_filter_set,
test_plan_live_matches_replay,
test_initial_cash_400k_large_buy,
test_paper_apply_buy_insufficient,
test_apply_buy_insufficient_cash,
]
for fn in tests:
try:

1
scripts/verify_env.py
View File

@@ -57,7 +57,6 @@ CONFIG_ENV_KEYS = {
"ALIGN_BB_POS_LOW",
"ALIGN_BB_POS_HIGH",
"DOWNLOAD_MONTHS",
"DOWNLOAD_MONTHS_1M",
"INCREMENTAL_OVERLAP_BARS",
"DOWNLOAD_BACKFILL_EXTRA_BARS",
"DOWNLOAD_MIN_INCREMENTAL_BARS",