"""TV 재구성용 읽기 전용 엔드포인트 테스트 (스키마·인증·빈 데이터·NULL net).""" import asyncio from datetime import date, datetime, timedelta from decimal import Decimal from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest from app.core.config import settings from app.core.security import require_api_key from app.services import tv_view_core as core from app.api.endpoints import tv_overlay class _Res: def __init__(self, rows): self._rows = rows def all(self): return self._rows def scalars(self): return SimpleNamespace(all=lambda: self._rows) class _DB: """execute 호출 순서대로 결과 반환 (SELECT 전용 페이크).""" def __init__(self, *results): self._results = list(results) async def execute(self, _q): return _Res(self._results.pop(0)) def _active(**kw): d = dict(stock_code="005930", stock_name="삼성전자", quantity=10, entry_price=Decimal("70000"), current_price=Decimal("72000"), stop_loss=None, take_profit=None, remaining_pct=Decimal("1.0"), highest_since_entry=None, strategy_name=None, created_at=datetime.now() - timedelta(minutes=90)) d.update(kw) return SimpleNamespace(**d) def _pf(code="005930", qty=10, cur=None): return SimpleNamespace(stock_code=code, quantity=qty, average_price=Decimal("70000"), current_price=Decimal(cur) if cur is not None else None) def _run(c): return asyncio.new_event_loop().run_until_complete(c) # ── 인증 ──────────────────────────────────────────────────────────────────── def test_all_new_routes_require_api_key(): from app.api import api_router want = {"/positions/overlay", "/risk/equity-history", "/strategy-config/display"} found = set() for r in api_router.routes: if r.path in want and "GET" in r.methods: deps = [d.call for d in r.dependant.dependencies] assert require_api_key in deps, r.path found.add(r.path) assert found == want def test_display_route_precedes_strategy_config_key_route(): from app.api import api_router paths = [r.path for r in api_router.routes if "GET" in getattr(r, "methods", ())] assert paths.index("/strategy-config/display") < paths.index("/strategy-config/{key}") # ── equity-history ────────────────────────────────────────────────────────── def test_equity_points_groups_and_null_net(): d1, d2 = date(2026, 10, 1), date(2026, 10, 2) rows = [(d1, "swing", Decimal("1000")), (d1, "daytrading", -300), (d1, "swing", None), (d2, "external", -2000), (d2, None, 50)] pts, nulls = core.build_equity_points(rows, 1_000_000) assert nulls == 1 assert pts[0]["realized_net_swing"] == 1000 and pts[0]["realized_net_dt"] == -300 assert pts[0]["cum_net"] == 700 and pts[0]["peak"] == 700 and pts[0]["drawdown_pct"] == 0 assert pts[1]["realized_net_other"] == -1950 and pts[1]["cum_net"] == -1250 assert pts[1]["drawdown_pct"] == pytest.approx((-1250 - 700) / 1_000_000 * 100, abs=1e-3) def test_equity_no_base_gives_null_drawdown(): pts, _ = core.build_equity_points([(date(2026, 10, 1), "swing", 5)], None) assert pts[0]["drawdown_pct"] is None def test_equity_endpoint_empty(): out = _run(tv_overlay.get_equity_history(days=90, db=_DB([]))) assert out["points"] == [] and out["null_net_rows"] == 0 and out["cost_basis"] == "net_after_costs" assert out["mdd_thresholds"] == {"reduce_pct": settings.MDD_PAUSE_THRESHOLD, "force_pct": settings.MDD_FORCE_THRESHOLD} def test_equity_endpoint_null_net_counted(): rows = [(datetime(2026, 10, 1, 10), "swing", None, "KRW"), (datetime(2026, 10, 1, 11), "swing", Decimal("10"), None), (datetime(2026, 10, 1, 12), "swing", Decimal("999"), "USD")] out = _run(tv_overlay.get_equity_history(days=90, db=_DB(rows))) assert out["null_net_rows"] == 1 and out["points"][0]["realized_net"] == 10 assert out["excluded_non_krw_rows"] == 1 # ── positions/overlay ─────────────────────────────────────────────────────── def _dt_state(positions): return patch("app.services.daytrading.position_manager.day_trading_state", SimpleNamespace(positions=positions)) def test_overlay_empty(): with _dt_state({}): out = _run(tv_overlay.get_positions_overlay(code=None, db=_DB([]))) assert out["positions"] == [] and "as_of" in out def test_overlay_swing_pct_basis_without_atr(): row = _active() with _dt_state({}), patch.object(tv_overlay, "_cached_atr", return_value=None): out = _run(tv_overlay.get_positions_overlay(code=None, db=_DB([row], [_pf()]))) p = out["positions"][0] assert p["strategy"] == "swing" and p["stop_basis"] == "pct" assert p["stop_loss_alt"] < p["stop_loss"] < 70000 and p["stop_note"] and p["pnl"] == 20000 and p["time_stop_at"] is None assert p["trailing_active"] is False and p["elapsed_min"] >= 89 def test_overlay_swing_atr_basis_and_stored_percent_ignored(): row = _active(stop_loss=Decimal("-7"), remaining_pct=Decimal("0.5"), highest_since_entry=Decimal("75000")) with _dt_state({}), patch.object(tv_overlay, "_cached_atr", return_value=1000.0): p = _run(tv_overlay.get_positions_overlay(code=None, db=_DB([row], [_pf()])))["positions"][0] assert p["stop_basis"] == "atr" and p["trailing_active"] is True and p["trailing_stop"] < 75000 def test_overlay_stored_price_used_when_no_atr_and_percent_value_not_trusted(): row = _active(stop_loss=Decimal("66000")) with _dt_state({}), patch.object(tv_overlay, "_cached_atr", return_value=None): p = _run(tv_overlay.get_positions_overlay(code=None, db=_DB([row], [_pf()])))["positions"][0] assert p["stop_basis"] == "stored" and p["stop_loss"] == 66000 assert core.plausible_stop_price(-7, 70000) is None and core.plausible_stop_price(7, 70000) is None def test_overlay_dt_from_memory(): et = datetime.now() - timedelta(minutes=10) pos = {"000660": {"entry_price": 100000.0, "quantity": 3, "current_price": 101000.0, "stop_loss": 98000.0, "target_price": float("inf"), "highest_price": 101500.0, "entry_time": et, "partial_sold": False, "stock_name": "SK하이닉스"}} with _dt_state(pos): out = _run(tv_overlay.get_positions_overlay(code="000660", db=_DB([]))) p = out["positions"][0] assert p["strategy"] == "dt" and p["target_price"] is None and p["highest_price"] == 101500.0 assert datetime.fromisoformat(p["time_stop_at"]) == et + timedelta(minutes=settings.DAY_TRADE_TIME_STOP_MINUTES) assert p["pnl"] == 3000 and p["stop_basis"] == "stored" # ── strategy-config/display ───────────────────────────────────────────────── def test_display_schema_and_settings_values(): out = _run(tv_overlay.get_strategy_display()) assert set(out) == {"rsi", "atr", "stop_loss_pct", "dt", "mdd", "adaptive_mode", "source", "sources"} assert out["source"] in ("adaptive", "strategy_cfg", "settings") assert out["stop_loss_pct"] < 0 assert out["dt"]["time_stop_min"] == settings.DAY_TRADE_TIME_STOP_MINUTES assert out["dt"]["eod_flat"] == f"{settings.DAY_TRADE_EOD_HOUR:02d}:{settings.DAY_TRADE_EOD_MIN:02d}" assert out["mdd"]["force_pct"] == settings.MDD_FORCE_THRESHOLD assert out["rsi"]["period"] == 14 or isinstance(out["rsi"]["period"], int) # ── trades 확장 / mdd thresholds ──────────────────────────────────────────── def test_trade_response_new_optional_fields(): from app.api.endpoints.trades import TradeHistoryResponse base = dict(id=1, rule_id=None, stock_code="005930", stock_name="x", order_type="SELL", order_price=None, order_quantity=1, executed_price=None, executed_quantity=None, total_amount=None, status="EXECUTED", order_id=None, message=None, entry_reason=None, exit_reason=None, signals=None, created_at=datetime.now(), executed_at=None) r = TradeHistoryResponse(**base) assert r.side == "sell" and r.net_profit_loss is None and r.commission is None and r.strategy_type is None assert TradeHistoryResponse(**{**base, "order_type": "BUY", "net_profit_loss": Decimal("5")}).side == "buy" def test_mdd_status_has_thresholds(): from app.api.endpoints.risk_management import _mdd_state_v2 t = _mdd_state_v2()["thresholds"] assert t == {"reduce_pct": settings.MDD_PAUSE_THRESHOLD, "force_pct": settings.MDD_FORCE_THRESHOLD} # ── 분봉 ──────────────────────────────────────────────────────────────────── def _m(hhmm, o, h, lo, c, v, ymd="20261016"): return {"stck_bsop_date": ymd, "stck_cntg_hour": hhmm + "00", "stck_oprc": str(o), "stck_hgpr": str(h), "stck_lwpr": str(lo), "stck_prpr": str(c), "cntg_vol": str(v)} def test_minute_candles_today_only_and_5m_aggregation(): today = date(2026, 10, 16) rows = [_m("0900", 10, 12, 9, 11, 5), _m("0901", 11, 15, 10, 14, 7), _m("0905", 14, 14, 13, 13, 1), _m("0902", 1, 1, 1, 1, 1, ymd="20261015"), _m("0903", 0, 0, 0, 0, 0)] c1 = core.minute_rows_to_candles(rows, today, 1) assert [x["volume"] for x in c1] == [5, 7, 1] c5 = core.minute_rows_to_candles(rows, today, 5) assert len(c5) == 2 and c5[0]["open"] == 10 and c5[0]["high"] == 15 and c5[0]["close"] == 14 and c5[0]["volume"] == 12 assert c5[1]["time"] - c5[0]["time"] == 300 assert core.minute_rows_to_candles([], today, 5) == [] def test_chart_endpoint_intraday_and_error_path(): from app.api.endpoints import market_data today = datetime.now(tz=None).date() market_data._INTRADAY_CACHE.clear() fake = MagicMock() fake.get_minute_chart_full_day = AsyncMock(return_value=[]) with patch.object(market_data, "korea_investment_api", fake): out = _run(market_data.get_stock_chart("005930", period="5m", realtime=False, limit=100)) assert out["data"] == [] and out["reason"] == "no_data" and out["period"] == "5m" fake.get_minute_chart_full_day = AsyncMock(side_effect=RuntimeError("boom")) market_data._INTRADAY_CACHE.clear() with patch.object(market_data, "korea_investment_api", fake): out = _run(market_data.get_stock_chart("005930", period="1m", realtime=False, limit=100)) assert out["reason"] == "chart_fetch_failed" def test_overlay_portfolio_price_preferred_one_row_per_code_and_orphan_excluded(): newer = _active(current_price=Decimal("1")) newer.id = 9 older = _active(entry_price=Decimal("50000")) older.id = 3 orphan = _active(stock_code="000660") orphan.id = 5 with _dt_state({}), patch.object(tv_overlay, "_cached_atr", return_value=None): out = _run(tv_overlay.get_positions_overlay(code=None, db=_DB([newer, orphan, older], [_pf(cur="73000")]))) assert [p["stock_code"] for p in out["positions"]] == ["005930"] assert out["positions"][0]["current_price"] == 73000 and out["positions"][0]["entry_price"] == 70000 assert out["as_of"].endswith("+09:00") def test_cached_atr_requires_fresh_and_after_entry(): import time as _t from app.services.swing import exit_service as ex now = datetime.now() ex._last_atr["T1"] = (500.0, _t.monotonic(), now) ex._last_atr["T2"] = (500.0, _t.monotonic() - settings.SWING_FALLBACK_PRICE_MAX_AGE_SEC - 5, now) try: assert tv_overlay._cached_atr("T1", now - timedelta(minutes=5)) == 500.0 assert tv_overlay._cached_atr("T1", now + timedelta(minutes=5)) is None # 진입 이전 값 assert tv_overlay._cached_atr("T2", now - timedelta(minutes=5)) is None # 오래됨 finally: ex._last_atr.pop("T1", None) ex._last_atr.pop("T2", None) def test_display_exit_rsi_matches_exit_logic_helper_and_adaptive(): from app.services import mean_reversion_strategy as m adp = SimpleNamespace(fallback_used=False, rsi_partial_exit=51.0, rsi_full_exit=66.0, rsi_entry=44.0, bear_rsi_entry=31.0, trading_mode="defensive") with patch("app.services.adaptive_strategy.get_current_adaptive_params", return_value=adp): assert m.resolve_swing_exit_rsi_v2() == (51.0, 66.0, True) out = _run(tv_overlay.get_strategy_display()) assert out["rsi"]["partial_exit"] == 51.0 and out["rsi"]["full_exit"] == 66.0 assert out["sources"] == {"entry": "adaptive", "exit": "adaptive"} and out["source"] == "adaptive" adp.fallback_used = True with patch("app.services.adaptive_strategy.get_current_adaptive_params", return_value=adp): assert m.resolve_swing_exit_rsi_v2() == (settings.MEAN_REVERSION_RSI_EXIT_THRESHOLD, settings.MEAN_REVERSION_RSI_FULL_EXIT, False) def test_exit_signal_uses_adaptive_rsi_partial_unchanged(): """check_exit_signal 이 헬퍼 경유로도 적응형 1차 익절 RSI 를 그대로 쓴다(동작 무변경).""" import inspect from app.services import mean_reversion_strategy as m src = inspect.getsource(m.check_exit_signal) assert "resolve_swing_exit_rsi_v2()" in src and "_adp_exit2" not in src def test_trade_response_numeric_floats(): from app.api.endpoints.trades import TradeHistoryResponse r = TradeHistoryResponse(id=1, rule_id=None, stock_code="x", stock_name=None, order_type="SELL", order_price=None, order_quantity=1, executed_price=None, executed_quantity=None, total_amount=None, status="EXECUTED", order_id=None, message=None, entry_reason=None, exit_reason=None, signals=None, created_at=datetime.now(), executed_at=None, commission=Decimal("12.5"), net_profit_loss=Decimal("-3")) assert isinstance(r.commission, float) and isinstance(r.net_profit_loss, float) def test_mdd_thresholds_equal_actual_transition_values(): """표시 임계 == mdd_manager 가 실제로 전이하는 값(-10/-20).""" from app.services.mdd_manager import MDDManager t = tv_overlay.mdd_thresholds() m = MDDManager() _run(m.update(1_000_000)) assert _run(m.update(1_000_000 * (1 + (t["reduce_pct"] - 0.1) / 100)))["action"] == "REDUCE_AND_PAUSE" assert _run(m.update(1_000_000 * (1 + (t["force_pct"] - 0.1) / 100)))["action"] == "FORCE_REDUCE" m2 = MDDManager() _run(m2.update(1_000_000)) assert _run(m2.update(1_000_000 * (1 + (t["reduce_pct"] + 0.5) / 100)))["action"] != "REDUCE_AND_PAUSE" def test_intraday_cache_and_inflight_share(): from app.api.endpoints import market_data market_data._INTRADAY_CACHE.clear() fake = MagicMock() fake.get_minute_chart_full_day = AsyncMock(return_value=[]) async def both(): a, b = await asyncio.gather(market_data.get_stock_chart("005930", period="1m", realtime=False, limit=100), market_data.get_stock_chart("005930", period="1m", realtime=False, limit=100)) c = await market_data.get_stock_chart("005930", period="1m", realtime=False, limit=100) return a, b, c with patch.object(market_data, "korea_investment_api", fake): a, b, c = _run(both()) assert a["reason"] == "no_data" == c["reason"] assert fake.get_minute_chart_full_day.await_count == 1