gahusb
7ea1a21487
Atomic mv of root V1 assets (main_server.py + modules/ + data/ + tests/ + entry scripts + docs + logs) into signal_v1/ subdirectory. load_dotenv() updated to load web-ai/.env explicitly via Path. Adds web-ai/CLAUDE.md (workspace guide) and web-ai/start.bat (signal_v1 entry wrapper). Prepares for signal_v2/ Phase 2. Tests: signal_v1/tests/unit baseline preserved (no regression). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
131 lines
5.7 KiB
Python
131 lines
5.7 KiB
Python
"""
|
||
일일 거래 장부 (DailyLedger) — v3.2
|
||
|
||
bot.py에 흩어져 있던 당일 상태를 한 객체로 집약:
|
||
- 당일 누적 매수금액 (KIS T+2 미차감 보완용)
|
||
- 연속 손절 카운터 + 매수 일시중단 타이머
|
||
- 미매도 종목의 매수 신호 점수 (앙상블 학습용)
|
||
- 일별 스냅샷/주간평가 플래그
|
||
|
||
날짜가 바뀌면 reset_if_new_day()가 자동 초기화.
|
||
순수 객체로 구현 — 외부 I/O 없음 → 단위 테스트 가능.
|
||
"""
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timedelta, date as date_cls
|
||
from typing import Dict, Optional
|
||
|
||
|
||
@dataclass
|
||
class DailyLedger:
|
||
# ── 당일 매수 회계 ──
|
||
today_buy_total: int = 0
|
||
today_buy_date: Optional[date_cls] = None
|
||
|
||
# ── 연속 손절 / 매수 일시 중단 ──
|
||
consecutive_stop_losses: int = 0
|
||
buy_paused_until: Optional[datetime] = None
|
||
stop_loss_pause_threshold: int = 3
|
||
stop_loss_pause_minutes: int = 30
|
||
|
||
# ── 앙상블 학습용: 미매도 종목의 매수 신호 점수 ──
|
||
buy_scores: Dict[str, dict] = field(default_factory=dict)
|
||
|
||
# ── 일일 플래그 ──
|
||
snapshot_taken: bool = False
|
||
weekly_eval_sent: bool = False
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 날짜 전환
|
||
# ──────────────────────────────────────────────
|
||
def reset_if_new_day(self, now: datetime) -> bool:
|
||
"""
|
||
오늘 날짜 기준으로 상태 초기화. 이미 오늘 자로 초기화됐으면 no-op.
|
||
|
||
Returns:
|
||
True — 실제로 초기화를 수행한 경우
|
||
False — 같은 날이라 그대로 둔 경우
|
||
"""
|
||
today = now.date()
|
||
if self.today_buy_date == today:
|
||
return False
|
||
self.today_buy_total = 0
|
||
self.today_buy_date = today
|
||
self.buy_scores.clear()
|
||
self.snapshot_taken = False
|
||
self.weekly_eval_sent = False
|
||
# 연속 손절 카운터 / 일시중단 타이머는 날짜 전환 시에만 초기화
|
||
self.consecutive_stop_losses = 0
|
||
self.buy_paused_until = None
|
||
return True
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 매수 / 매도 기록
|
||
# ──────────────────────────────────────────────
|
||
def record_buy(self, ticker: str, amount: int, scores: dict) -> None:
|
||
"""매수 체결 기록. amount는 집행 금액(원), scores는 앙상블 신호."""
|
||
self.today_buy_total += int(amount)
|
||
self.buy_scores[ticker] = dict(scores)
|
||
|
||
def pop_buy_scores(self, ticker: str) -> Optional[dict]:
|
||
"""매도 체결 시 앙상블 학습을 위해 매수 당시 신호를 반환하고 제거."""
|
||
return self.buy_scores.pop(ticker, None)
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 손절 관리
|
||
# ──────────────────────────────────────────────
|
||
def record_sell_outcome(self, outcome_pct: float, now: datetime) -> bool:
|
||
"""
|
||
매도 결과를 반영해 연속 손절 카운터 업데이트.
|
||
|
||
Returns:
|
||
True — 임계치 도달 → 매수 일시중단 활성화됨
|
||
False — 임계치 미도달
|
||
"""
|
||
if outcome_pct < 0:
|
||
self.consecutive_stop_losses += 1
|
||
if self.consecutive_stop_losses >= self.stop_loss_pause_threshold:
|
||
self.buy_paused_until = now + timedelta(
|
||
minutes=self.stop_loss_pause_minutes
|
||
)
|
||
return True
|
||
else:
|
||
self.consecutive_stop_losses = 0
|
||
return False
|
||
|
||
def is_buy_paused(self, now: datetime) -> bool:
|
||
"""
|
||
매수 일시중단 상태 조회. 만료되면 자동 해제 + 카운터 리셋.
|
||
"""
|
||
if self.buy_paused_until is None:
|
||
return False
|
||
if now >= self.buy_paused_until:
|
||
self.buy_paused_until = None
|
||
self.consecutive_stop_losses = 0
|
||
return False
|
||
return True
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 예수금 계산 (KIS T+2 보완)
|
||
# ──────────────────────────────────────────────
|
||
def effective_today_buy(self, kis_today_buy: int) -> int:
|
||
"""
|
||
KIS API가 반환한 당일 매수금(`thdt_buy_amt`)과
|
||
로컬 누적값 중 더 큰 값을 신뢰.
|
||
(모의투자는 T+2 미차감으로 인해 과소 보고되는 경우 있음)
|
||
"""
|
||
return max(int(kis_today_buy or 0), self.today_buy_total)
|
||
|
||
def available_deposit(self, raw_deposit: int, max_daily_buy_ratio: float,
|
||
kis_today_buy: int = 0) -> int:
|
||
"""
|
||
당일 사용 가능한 예수금 계산.
|
||
|
||
max_daily_buy = raw_deposit × ratio
|
||
avail = min(raw_deposit, max_daily_buy) − effective_today_buy
|
||
"""
|
||
if raw_deposit <= 0:
|
||
return 0
|
||
max_daily_buy = int(raw_deposit * max_daily_buy_ratio)
|
||
used = self.effective_today_buy(kis_today_buy)
|
||
return max(0, min(raw_deposit, max_daily_buy) - used)
|