feat: 데이트레이딩 ORB 복구 + 스윙:DT 자본 50:50 + 시간대 제한 완화

- ORB: 09:00에 당일 분석이 없으면 최근 분석일 결과로 watchlist 선채움 → ORB 수집 가능.
  자동 목록은 30분마다 갱신(당일 분석 완료 시 교체), 수동 설정 목록은 덮어쓰지 않음.
- 자본 배분: DAY_TRADE_MAX_RATIO_DT 0.25→0.50, 종목당 4%→6%, 동시 3→8종목.
  SWING_MAX_CAPITAL_RATIO=0.50 신설 — 스윙 합산 투자액 초과 시 스윙 신규 매수 차단(trade_executor).
- 시간대: 점심(11:30~13:00) VWAP/RSI_BB 제한 제거, 하드코딩 14:30 마감 → 설정값 15:00.
- 회귀 테스트 3건, CLAUDE.md §6/§7.3/§8.2 갱신.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-22 13:29:02 +09:00
parent 62c324e346
commit e0178aaf0d
6 changed files with 196 additions and 28 deletions

View File

@@ -0,0 +1,111 @@
"""2026-09-22 변경 회귀 테스트: ORB 감시목록 전일 폴백 / 시간대 제한 완화 / 스윙:DT 50:50 자본 배분."""
import asyncio
from datetime import date, datetime, timedelta, time as dtime
from unittest.mock import AsyncMock, patch
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.compiler import compiles
from app.core.config import settings
@compiles(JSONB, "sqlite")
def _jsonb_as_json(type_, compiler, **kw): # 테스트용 SQLite 호환
return "JSON"
from app.models.models import Base, StockAnalysis
from app.services.daytrading import entry_service as es
from app.services.daytrading import scheduler as sch
async def _mk_session():
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
async with engine.begin() as conn:
await conn.run_sync(lambda c: Base.metadata.create_all(c, tables=[StockAnalysis.__table__]))
return engine, async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
def _row(code, d, score=3, price=10000):
return StockAnalysis(stock_code=code, stock_name=f"N{code}", analysis_date=d, score=score,
rsi=40.0, current_price=price)
@pytest.mark.asyncio
async def test_populate_falls_back_to_latest_analysis_date():
engine, Session = await _mk_session()
today = datetime.now(es._KST).date()
yday = today - timedelta(days=1)
async with Session() as s:
s.add_all([_row("000001", yday), _row("000002", yday, score=5), _row("000003", yday, price=3000)])
await s.commit()
with patch("app.core.database.AsyncSessionLocal", Session):
n = await es.populate_daytrading_watchlist()
assert n == 2 # 5,000원 미만 제외
assert set(es.get_watchlist()) == {"000001", "000002"}
assert es.is_watchlist_auto() is True
# 당일 데이터가 생기면 당일 결과가 우선
async with Session() as s:
s.add(_row("000009", today)); await s.commit()
with patch("app.core.database.AsyncSessionLocal", Session):
n = await es.populate_daytrading_watchlist()
assert n == 1 and set(es.get_watchlist()) == {"000009"}
# 수동 설정 후에는 auto 플래그 해제 → 스케줄러 30분 갱신 대상 아님
es.set_watchlist([{"code": "005930", "name": "삼성전자"}])
assert es.is_watchlist_auto() is False
await engine.dispose()
@pytest.mark.asyncio
async def test_entry_no_lunch_block_and_cutoff_from_settings():
"""점심시간(12:00)에도 VWAP/RSI_BB 스캔이 수행되고, 마감은 설정값(15:00) 기준."""
from app.services.daytrading.position_manager import day_trading_state
es.set_watchlist([{"code": "005930", "name": "삼성전자"}])
captured = {}
async def fake_try_enter(code, name, total_assets, available_cash, is_lunch_break=False):
captured["is_lunch_break"] = is_lunch_break
return None
class _Adp: # 적응형: 데이트레이딩 허용
fallback_used = False; daytrading_enabled = True; trading_mode = "normal"
async def run_at(hh, mm):
fake_now = datetime.now(es._KST).replace(hour=hh, minute=mm, second=0)
class _DT(datetime):
@classmethod
def now(cls, tz=None):
return fake_now if tz else fake_now.replace(tzinfo=None)
with patch.object(es, "datetime", _DT), \
patch.object(es.mdd_manager, "is_short_term_strategy_enabled", AsyncMock(return_value=True)), \
patch("app.services.adaptive_strategy.get_current_adaptive_params", return_value=_Adp()), \
patch.object(es, "is_daytrading_allowed_by_playbook", return_value=True), \
patch.object(es, "check_market_regime", AsyncMock(return_value={"regime": "neutral"})), \
patch.object(es.korea_investment_api, "get_balance", AsyncMock(return_value={"output2": [{"tot_evlu_amt": "10000000", "prvs_rcdl_excc_amt": "5000000"}]})), \
patch.object(es, "_try_enter", fake_try_enter):
captured.clear()
return await es.scan_and_enter()
day_trading_state.positions.clear()
await run_at(12, 0)
assert captured.get("is_lunch_break") is False, "점심시간 제한이 제거되어야 함"
await run_at(14, 45)
assert captured.get("is_lunch_break") is False, "14:30~15:00 구간도 진입 스캔해야 함"
await run_at(15, 5)
assert captured == {}, "15:00 이후에는 스캔하지 않아야 함"
def test_capital_split_settings():
assert settings.DAY_TRADE_MAX_RATIO_DT == 0.50
assert settings.SWING_MAX_CAPITAL_RATIO == 0.50
assert settings.DAY_TRADE_MAX_POSITIONS * settings.DAY_TRADE_MAX_CAPITAL_RATIO >= 0.48
assert (settings.DAY_TRADE_ENTRY_CUTOFF_HOUR, settings.DAY_TRADE_ENTRY_CUTOFF_MIN) == (15, 0)
from app.services.day_trading_strategy import calculate_day_trade_position_size
# 총자산 1,000만: 종목당 60만, DT 합산 500만
assert calculate_day_trade_position_size(10_000_000, 5_000_000, 0) == 600_000
assert calculate_day_trade_position_size(10_000_000, 5_000_000, 4_800_000) == 200_000