shuim
a264561576
- 스윙 체결가 확인 전 executed_price NULL + fill_sweep, 진입가 resolve(ActiveTrade↔Portfolio), 0체결/부분체결 매수 장부 원복 - DT 고아 감지(근거 한정)·0체결 매도 복원·stale 점검 확대, 시간손절 -0.3% 원복 - 폴백 손절 ATR 캐시, 거래량부족-차단 신호 매수 차단, 현금 5% 유보 재적용, MDD peak/reset 수정 - KIS 서킷브레이커 청산 우회·경보, db_worker 재큐잉 순서, heartbeat·health-watch 사이드카 - 실시간 WS 재구독·41건 우선순위·다건 프레임 분할, 차트 샘플 제거·당일봉 합성, WS 서브프로토콜 인증 - 프론트: 미관리 그룹·확인 모달·차트 실시간 반영·예수금 표시 정정 - 전문가 에이전트 6종 추가(.claude/agents) - 유닛 스위트 2667 passed (소켓 차단 하), tsc OK Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y83Za98XsNw9zuA4FTUytN
94 lines
3.4 KiB
Python
94 lines
3.4 KiB
Python
"""kr_backtest_engine: 결측일 가짜 봉(is_filler)에서는 체결·신호·지표를 쓰지 않는다."""
|
|
from datetime import date, timedelta
|
|
|
|
from app.services import kr_backtest_engine as eng
|
|
from app.services.kr_backtest_engine import (
|
|
BarData, BacktestStrategy, KrBacktestEngine, MeanReversionBacktest,
|
|
_align_bars_to_dates,
|
|
)
|
|
|
|
D0 = date(2026, 1, 5)
|
|
|
|
|
|
def _dates(n):
|
|
return [D0 + timedelta(days=i) for i in range(n)]
|
|
|
|
|
|
def _bar(d, px, vol=1000, code="A"):
|
|
return BarData(trade_date=d, open_price=px, high_price=px + 1, low_price=px - 1,
|
|
close_price=px, volume=vol, stock_code=code)
|
|
|
|
|
|
def test_align_marks_filler_and_keeps_real():
|
|
ds = _dates(4)
|
|
out = _align_bars_to_dates({"A": [_bar(ds[0], 100), _bar(ds[3], 90)]}, ds)["A"]
|
|
assert [b.is_filler for b in out] == [False, True, True, False]
|
|
assert out[1].volume == 0
|
|
|
|
|
|
class _ScriptedStrategy(BacktestStrategy):
|
|
"""지정한 bar_idx 에서 주문을 낸다."""
|
|
def __init__(self, script):
|
|
self.script = script
|
|
|
|
def name(self): return "s"
|
|
def description(self): return "s"
|
|
def default_params(self): return {}
|
|
|
|
def on_bar(self, bar_idx, bars, positions, cash, total_equity, params):
|
|
return list(self.script.get(bar_idx, []))
|
|
|
|
|
|
def _buy(q=10):
|
|
return {"action": "BUY", "stock_code": "A", "price": 100, "quantity": q, "reason": "t", "sell_pct": 0.0}
|
|
|
|
|
|
def _sell(q=10):
|
|
return {"action": "SELL", "stock_code": "A", "price": 100, "quantity": q, "reason": "t", "sell_pct": 1.0}
|
|
|
|
|
|
def _run(bars, ds, script):
|
|
st = _ScriptedStrategy(script)
|
|
return KrBacktestEngine()._simulate({"A": bars}, ds, st, {}, 1_000_000, "KOSPI", 1)
|
|
|
|
|
|
def test_sell_not_filled_on_filler_bar_carries_to_next_real_bar():
|
|
ds = _dates(5)
|
|
# 0,1:실 2,3:가짜(정지) 4:실(갭하락 50)
|
|
aligned = _align_bars_to_dates({"A": [_bar(ds[0], 100), _bar(ds[1], 100), _bar(ds[4], 50)]}, ds)["A"]
|
|
res = _run(aligned, ds, {0: [_buy()], 1: [_sell()]})
|
|
sells = [t for t in res.trades if t["side"] == "SELL"]
|
|
assert len(sells) == 1
|
|
# 정지일의 평평한 가격(100 근처)이 아니라 재개일 갭 시가(50 근처)에 체결
|
|
assert sells[0]["exit_date"] == str(ds[4])
|
|
assert sells[0]["exit_price"] < 60
|
|
|
|
|
|
def test_buy_signal_on_filler_next_bar_is_dropped():
|
|
ds = _dates(4)
|
|
aligned = _align_bars_to_dates({"A": [_bar(ds[0], 100), _bar(ds[1], 100), _bar(ds[3], 100)]}, ds)["A"]
|
|
res = _run(aligned, ds, {1: [_buy()]}) # ds[2] 는 가짜 봉
|
|
assert [t for t in res.trades if t["side"] == "BUY"] == []
|
|
|
|
|
|
def test_mean_reversion_on_bar_ignores_filler_days(monkeypatch):
|
|
n = 70
|
|
ds = _dates(n + 2)
|
|
real = [_bar(ds[i], 10000 + (i % 3)) for i in range(n)] + [_bar(ds[n + 1], 10000)]
|
|
aligned = _align_bars_to_dates({"A": real}, ds)
|
|
seen = []
|
|
|
|
def spy_rsi(closes, period=14):
|
|
seen.append(len(closes))
|
|
return 50.0
|
|
monkeypatch.setattr(eng, "calculate_rsi", spy_rsi)
|
|
monkeypatch.setattr(eng, "calculate_bollinger", lambda c, p=20: (110.0, 100.0, 90.0))
|
|
monkeypatch.setattr(eng, "check_buy_signal_backtest_v2", lambda *a, **k: None)
|
|
s = MeanReversionBacktest()
|
|
p = s.default_params()
|
|
assert aligned["A"][n].is_filler
|
|
assert s.on_bar(n, aligned, {}, 1e6, 1e6, p) == []
|
|
assert seen == [] # 가짜 봉 날에는 지표 계산 자체가 없다
|
|
s.on_bar(n + 1, aligned, {}, 1e6, 1e6, p)
|
|
assert seen and seen[0] == n + 1 # 가짜 봉 1개가 입력에서 제외됨
|