e1ac8d119b
- 프론트엔드 UI 업데이트 (backtest.html, backtest.js) 엔진 라디오 버튼 통합 관련 반영 - Rust 플러그인(kis_rust_core) 및 컴파일 소스코드 추가 - CLI 백테스트 스크립트 수정 및 최신화 - 기타 스크래치 테스트 스크립트, 로그 요약 마크다운(.md) 등 누락 파일 일괄 반영 - 추가적으로 아직 발견되지 않은 엣지 케이스나 렌더링 오류가 포함되어 있을 가능성이 있음
120 lines
4.3 KiB
Python
120 lines
4.3 KiB
Python
import sys
|
|
import os
|
|
import time
|
|
import datetime
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from database import TradeDB
|
|
from kis_trader.engine.feed_fallback import packet_lag_seconds, is_feed_read_stale, bar_end_datetime, tick_in_bar_bucket
|
|
|
|
def old_bar_is_garbage(ticks, candle_time, tf_min=1):
|
|
bucket = []
|
|
for t in ticks:
|
|
raw = str(t.get("tick_time_raw") or t.get("tick_time") or "")
|
|
if tick_in_bar_bucket(str(raw), candle_time, tf_min):
|
|
bucket.append(t)
|
|
|
|
if not bucket:
|
|
return True, "No ticks"
|
|
|
|
bar_end = bar_end_datetime(candle_time, tf_min)
|
|
if bar_end is None:
|
|
return False, "No bar_end"
|
|
|
|
stale_lags = []
|
|
for t in bucket:
|
|
raw = str(t.get("tick_time_raw") or t.get("tick_time") or "")
|
|
lag = packet_lag_seconds(raw, now_dt=bar_end)
|
|
stale = is_feed_read_stale(lag)
|
|
if not stale:
|
|
return False, f"Valid tick lag: {lag:.2f}s"
|
|
stale_lags.append(lag)
|
|
|
|
return True, f"All ticks stale relative to bar_end (min lag={min(stale_lags):.2f}s)"
|
|
|
|
def main():
|
|
db = TradeDB()
|
|
# 동국알앤에스 (323350) 2026-09-04
|
|
code = "323350"
|
|
day = "20260904"
|
|
|
|
print(f"[{code}] {day} 캔들 조회 중...")
|
|
candles = db.conn.execute(
|
|
"SELECT * FROM ws_candles WHERE code=%s AND candle_time LIKE %s AND timeframe=1 ORDER BY candle_time ASC",
|
|
(code, f"{day}%%")
|
|
).fetchall()
|
|
|
|
candles = [dict(c) for c in candles]
|
|
print(f"총 {len(candles)}개 1분봉 캔들 확보.\n")
|
|
|
|
print(f"[{code}] {day} 틱 조회 중...")
|
|
ticks = db.conn.execute(
|
|
"SELECT * FROM ws_ticks WHERE code=%s AND tick_time LIKE %s ORDER BY tick_time ASC",
|
|
(code, f"{day}%%")
|
|
).fetchall()
|
|
|
|
ticks_dict = []
|
|
for t in ticks:
|
|
ticks_dict.append({
|
|
"tick_time": t["tick_time"],
|
|
"tick_time_raw": t["tick_time"],
|
|
"price": t["price"],
|
|
"volume": t["volume"]
|
|
})
|
|
print(f"총 {len(ticks_dict)}개 틱 확보.\n")
|
|
|
|
if not ticks_dict:
|
|
print("틱 데이터가 없어 검증을 종료합니다.")
|
|
return
|
|
|
|
garbage_count = 0
|
|
valid_count = 0
|
|
empty_count = 0
|
|
|
|
print("--- 🗑️ 기존 로직에 의해 '쓰레기(Garbage)'로 판정된 정상 캔들 샘플 ---\n")
|
|
|
|
for c in candles:
|
|
# ws_candles는 candle_time, ls_ws_candles는 datetime 컬럼을 가짐
|
|
ct = c.get("candle_time") or c.get("datetime")
|
|
if not ct: continue
|
|
bucket_ticks = [t for t in ticks_dict if tick_in_bar_bucket(t["tick_time"], ct, 1)]
|
|
|
|
is_garbage, reason = old_bar_is_garbage(bucket_ticks, ct, 1)
|
|
|
|
if len(bucket_ticks) == 0:
|
|
empty_count += 1
|
|
continue
|
|
|
|
if is_garbage:
|
|
garbage_count += 1
|
|
if garbage_count <= 5: # 5개만 샘플 출력
|
|
vol = sum(t["volume"] for t in bucket_ticks)
|
|
last_tick_time = bucket_ticks[-1]["tick_time"]
|
|
bar_end_time = bar_end_datetime(ct, 1)
|
|
|
|
print(f"[Garbage 오판] 캔들시간: {ct}")
|
|
print(f" -> 분봉 내 체결 틱 수: {len(bucket_ticks)}개")
|
|
print(f" -> 분봉 내 거래량: {vol}주")
|
|
print(f" -> 마지막 체결 틱 시간: {last_tick_time}")
|
|
print(f" -> 분봉 종료 정각 시간: {bar_end_time.strftime('%Y%m%d%H%M%S')}")
|
|
print(f" -> 판정 이유: {reason}")
|
|
print(f" => 진짜 쓰레기? ❌ 정상 캔들인데 마지막 체결이 정각보다 몇초 일찍 끝났을 뿐입니다.\n")
|
|
else:
|
|
valid_count += 1
|
|
|
|
print(f"=== 📊 최종 결과 요약 ===")
|
|
print(f"총 캔들 수: {len(candles)}")
|
|
print(f"거래가 없었던 빈 캔들: {empty_count}")
|
|
print(f"정상 판정(통과) 캔들: {valid_count}")
|
|
print(f"가비지로 오판(삭제)된 캔들: {garbage_count}")
|
|
|
|
if (valid_count + garbage_count) > 0:
|
|
drop_rate = (garbage_count / (valid_count + garbage_count)) * 100
|
|
print(f"🚀 거래가 있었던 캔들 중 멀쩡한데 버려진 비율: {drop_rate:.1f}%")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|