fc27e726f9
변경 사항 (Changes): 구문 오류(Syntax error) 및 토큰 낭비를 방지하기 위해 에이전트 쉘(Agent shell)과 파이썬 코드 스니펫에 다수의 신규 안전 규칙(Safety rules)을 추가함. 스키마 검증 및 적절한 SQL 포맷팅을 보장하기 위해 임시(Ad-hoc) 데이터베이스 쿼리 작성 가이드라인을 도입함. 코드 수정 후 UI 기능이 정상 작동하는지 확인하기 위해, 백테스트 웹 서비스 재시작 및 브라우저 검증에 대한 새로운 규칙을 구현함. 시스템 전반의 무결성(Integrity)을 유지하기 위해 실전 매매(Live trading), 웹 백테스팅, 파라미터 탐색(Parameter searches) 간의 일관성 검사(Consistency checks) 체계를 확립함. 기대 효과 (Impact): 이러한 개선 사항들은 트레이딩 시스템의 견고성(Robustness)과 신뢰성을 향상시키며, 에러 발생을 최소화하고 다양한 시스템 컴포넌트 간의 원활한 상호작용을 보장함.
107 lines
3.3 KiB
Python
107 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
env 타임라인 — 백테스트 시 env_config / config_* 변경 시각별 스냅샷 재생.
|
|
|
|
실매·파람서치는 최신 env 1행만 사용. 웹 백테만 ``EnvTimeline`` 으로
|
|
봉 시각(candle_time)마다 그 시점까지의 DB 설정을 적용한다.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from bisect import bisect_right
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
|
|
def _created_at_to_key(created_at: str) -> str:
|
|
"""``YYYY-MM-DD HH:MM:SS`` → ``YYYYMMDDHHMMSS``."""
|
|
s = str(created_at or "").strip()
|
|
if len(s) >= 19:
|
|
return (
|
|
s[0:4] + s[5:7] + s[8:10]
|
|
+ s[11:13] + s[14:16] + s[17:19]
|
|
)
|
|
return ""
|
|
|
|
|
|
def _candle_time_to_lookup_key(candle_time: str) -> str:
|
|
"""분봉 키(YYYYMMDDHHMM) → 해당 분 마감초(…59) lookup 키."""
|
|
t = str(candle_time or "")[:12]
|
|
if len(t) < 12:
|
|
return ""
|
|
return t + "59"
|
|
|
|
|
|
def _ymd_to_start_bound(ymd: str) -> str:
|
|
y = str(ymd or "")[:8]
|
|
if len(y) < 8:
|
|
return ""
|
|
return f"{y[:4]}-{y[4:6]}-{y[6:8]} 00:00:00"
|
|
|
|
|
|
class EnvTimeline:
|
|
"""env 변경 이벤트 시계열 — ``snapshot_at`` 으로 봉 시각별 merged env 조회."""
|
|
|
|
def __init__(self, events: List[Tuple[str, Dict[str, Any]]]) -> None:
|
|
deduped: List[Tuple[str, Dict[str, Any]]] = []
|
|
prev_sig: Optional[Tuple[str, ...]] = None
|
|
for key, snap in events:
|
|
if not key:
|
|
continue
|
|
sig = tuple(sorted((k, str(v)) for k, v in snap.items()))
|
|
if sig == prev_sig:
|
|
continue
|
|
prev_sig = sig
|
|
deduped.append((key, dict(snap)))
|
|
self._keys: List[str] = [e[0] for e in deduped]
|
|
self._snaps: List[Dict[str, Any]] = [e[1] for e in deduped]
|
|
|
|
@property
|
|
def snapshot_count(self) -> int:
|
|
return len(self._keys)
|
|
|
|
def snapshot_at(self, candle_time: str) -> Tuple[Dict[str, Any], str]:
|
|
"""봉 시각 기준 적용 env 스냅샷 + 이벤트 키(YYYYMMDDHHMMSS)."""
|
|
if not self._keys:
|
|
return {}, ""
|
|
qkey = _candle_time_to_lookup_key(candle_time)
|
|
if not qkey:
|
|
return dict(self._snaps[0]), self._keys[0]
|
|
idx = bisect_right(self._keys, qkey) - 1
|
|
if idx < 0:
|
|
return {}, ""
|
|
return dict(self._snaps[idx]), self._keys[idx]
|
|
|
|
|
|
def build_env_timeline(
|
|
db: Any,
|
|
start_ymd: str,
|
|
end_ymd: str,
|
|
) -> Optional[EnvTimeline]:
|
|
"""백테 구간 env 변경 이력 → ``EnvTimeline`` (변경 없으면 1스냅샷)."""
|
|
if db is None:
|
|
return None
|
|
sy = str(start_ymd or "")[:8]
|
|
ey = str(end_ymd or "")[:8]
|
|
if len(sy) < 8 or len(ey) < 8:
|
|
return None
|
|
if not hasattr(db, "list_env_change_times_in_range"):
|
|
return None
|
|
if not hasattr(db, "get_merged_env_snapshot_as_of"):
|
|
return None
|
|
|
|
start_bound = _ymd_to_start_bound(sy)
|
|
event_times = db.list_env_change_times_in_range(sy, ey)
|
|
if not event_times:
|
|
event_times = [start_bound]
|
|
|
|
events: List[Tuple[str, Dict[str, Any]]] = []
|
|
for as_of in event_times:
|
|
merged = db.get_merged_env_snapshot_as_of(as_of)
|
|
key = _created_at_to_key(as_of)
|
|
if not key:
|
|
continue
|
|
events.append((key, merged))
|
|
|
|
if not events:
|
|
return None
|
|
return EnvTimeline(events)
|