"""2026-10-08 체결가/평단/환율/총자산 정합 회귀. 1) 체결 확인 전 executed_price 에 주문가를 넣지 않고, 확인 시 체결가×체결수량·순손익으로 갱신 2) 스윙 손익·손절 진입가 = 스윙 ActiveTrade.entry_price (없으면 Portfolio 평단 폴백, 괴리 시 보수적 max) 3) 환율 0 → 마지막 성공값/설정값 폴백, 모두 없으면 환산 생략 플래그 4) 총자산 = 주식평가 + D+2 예수금 분해 필드 """ import asyncio from decimal import Decimal from types import SimpleNamespace from unittest.mock import AsyncMock import pytest from app.core.config import settings import app.services.db_worker as dbw import app.services.trade_executor as te from app.services.swing import fill_sweep as fs from app.services.swing.position_manager import ( resolve_swing_entry_price_v2, SWING_ENTRY_PRICE_MAX_DEVIATION, ) from app.api.endpoints import portfolio as pf def _run(coro): return asyncio.run(coro) # ── 1. 체결 기준 TradeHistory 필드 ──────────────────────────────────────────── def test_fill_fields_buy_amount_is_fill_price_times_qty(): f = te._fill_update_fields_v2("BUY", 9, 41200.0, -0.74, None) assert f["executed_price"] == Decimal("41200.0") assert f["executed_quantity"] == 9 assert f["total_amount"] == Decimal(str(round(41200.0 * 9, 2))) assert "net_profit_loss" not in f def test_fill_fields_no_price_never_substitutes_order_price(): f = te._fill_update_fields_v2("BUY", 9, 0.0, 0.0, None) assert "executed_price" not in f assert "total_amount" not in f assert f["executed_quantity"] == 9 def test_fill_fields_sell_recomputes_net_from_fill(): basis = {"entry_price": 10000.0, "market": "KOSPI", "is_domestic": True, "net_rate": False} f = te._fill_update_fields_v2("SELL", 10, 11000.0, 0.0, basis) from app.services.trade_helpers import calculate_trade_costs c = calculate_trade_costs(entry_price=10000.0, exit_price=11000.0, quantity=10, is_domestic=True, market="KOSPI") assert float(f["net_profit_loss"]) == pytest.approx(10000.0 - float(c["total_cost"]), abs=0.01) assert float(f["profit_loss_rate"]) == pytest.approx(10.0, abs=1e-6) # gross 규약 assert float(f["commission"]) == pytest.approx(float(c["total_cost"])) def test_fill_fields_sell_net_rate(): basis = {"entry_price": 10000.0, "net_rate": True} f = te._fill_update_fields_v2("SELL", 10, 11000.0, 0.0, basis) assert float(f["profit_loss_rate"]) == pytest.approx( float(f["net_profit_loss"]) / 100000.0 * 100, abs=1e-3) @pytest.fixture def verify_env(monkeypatch): calls = {"th": [], "at": []} async def _th(order_id, fields): calls["th"].append((order_id, dict(fields))) async def _at(data): calls["at"].append(dict(data)) monkeypatch.setattr(dbw, "enqueue_update_trade_history_by_order_id", _th) monkeypatch.setattr(dbw, "enqueue_active_trade_upsert", _at) monkeypatch.setattr(te, "_delayed_sync_portfolio", AsyncMock()) monkeypatch.setattr(te.asyncio, "sleep", AsyncMock()) te._entry_fill_adjusted_orders.clear() return calls def _patch_exec(monkeypatch, result): """체결조회 목 — 테스트 시점에 두 참조(trade_executor 캡처본, 소스 모듈 속성)를 모두 덮는다. (다른 테스트 파일이 sys.modules 의 KIS 모듈을 MagicMock 으로 바꿔도 순서 무관하게 동작) """ import sys m = AsyncMock(return_value=result) stub = SimpleNamespace(get_order_execution=m) monkeypatch.setattr(te, "korea_investment_api", stub) mod = sys.modules.get("app.services.korea_investment_api") if mod is not None: monkeypatch.setattr(mod, "korea_investment_api", stub, raising=False) return m def test_verify_v3_buy_updates_amount_and_entry(monkeypatch, verify_env): _patch_exec(monkeypatch, {"ok": True, "found": True, "filled_qty": 9, "filled_price": 41200.0, "remaining_qty": 0}) _run(te._verify_and_sync_execution_v3( "0000123", "082740", "한화엔진", 9, 41506.0, "BUY", True, "KRX", 0, fill_entry_strategy_name="swing", sell_pnl_basis={"strategy_type": "swing"}, )) (oid, fields), = verify_env["th"] assert fields["executed_price"] == Decimal("41200.0") assert fields["total_amount"] == Decimal(str(41200.0 * 9)) assert fields["order_type"] == "BUY" and fields["strategy_type"] == "swing" (at,) = verify_env["at"] assert at["strategy_name"] == "swing" assert at["entry_fill_adjust"] == {"order_price": 41506.0, "filled_price": 41200.0, "filled_qty": 9} assert te.entry_fill_already_adjusted_v2("123") def test_verify_v3_missing_fill_price_no_price_written(monkeypatch, verify_env): _patch_exec(monkeypatch, {"ok": True, "found": True, "filled_qty": 9, "filled_price": 0, "remaining_qty": 0}) _run(te._verify_and_sync_execution_v3( "0000124", "005380", "현대차", 9, 337177.0, "BUY", True, "KRX", 0, fill_entry_strategy_name="swing", )) (oid, fields), = verify_env["th"] assert "executed_price" not in fields and "total_amount" not in fields assert verify_env["at"] == [] # 체결가 모르면 진입가 보정 없음 def test_verify_v3_sell_net_from_fill(monkeypatch, verify_env): _patch_exec(monkeypatch, {"ok": True, "found": True, "filled_qty": 5, "filled_price": 9500.0, "remaining_qty": 0}) _run(te._verify_and_sync_execution_v3( "0000125", "005930", "삼성", 5, 9600.0, "SELL", True, "KRX", 0, sell_pnl_basis={"entry_price": 10000.0, "market": "KOSPI", "strategy_type": "swing"}, )) (oid, fields), = verify_env["th"] assert float(fields["net_profit_loss"]) < -2500.0 # gross -2500 에서 비용 차감 assert float(fields["profit_loss_rate"]) == pytest.approx(-5.0) # ── 1-b. ActiveTrade 진입가 누적/체결 보정 (db_worker 순수 로직) ────────────── def _rec(**kw): base = dict(id=7, quantity=10, entry_price=10000.0, highest_since_entry=10500.0, remaining_pct=1.0) base.update(kw) return SimpleNamespace(**base) def test_accumulate_weighted_average_and_pinned_retry(): r = _rec() data = {"accumulate_entry": True, "quantity": 10, "entry_price": 8000.0, "highest_since_entry": 8000.0} skip = dbw._apply_entry_price_adjustments_v2(r, data) assert {"quantity", "entry_price", "highest_since_entry"} <= skip assert r.quantity == 20 and r.entry_price == 9000.0 assert r.highest_since_entry == 10500.0 # 고점은 낮추지 않는다 # 같은 payload 재실행(종료 drain) — 두 번 누적하지 않는다 r2 = _rec() dbw._apply_entry_price_adjustments_v2(r2, data) assert r2.quantity == 20 and r2.entry_price == 9000.0 def test_accumulate_after_partial_exit_weights_held_only(): r = _rec(remaining_pct=0.5) # 10주 중 5주 보유 dbw._apply_entry_price_adjustments_v2(r, {"accumulate_entry": True, "quantity": 5, "entry_price": 8000.0}) assert r.entry_price == 9000.0 # (5×10000 + 5×8000)/10 def test_accumulate_into_empty_row_takes_new_price(): r = _rec(quantity=0, entry_price=0) dbw._apply_entry_price_adjustments_v2(r, {"accumulate_entry": True, "quantity": 3, "entry_price": 5000.0}) assert r.quantity == 3 and r.entry_price == 5000.0 def test_fill_adjust_new_entry_becomes_fill_price(): r = _rec(quantity=9, entry_price=41506.0) d = {"entry_fill_adjust": {"order_price": 41506.0, "filled_price": 41200.0, "filled_qty": 9}} dbw._apply_entry_price_adjustments_v2(r, d) assert r.entry_price == 41200.0 # 재실행 시 고정값 재사용(두 번 빼지 않음) r.entry_price = 41506.0 dbw._apply_entry_price_adjustments_v2(r, d) assert r.entry_price == 41200.0 def test_fill_adjust_partial_of_accumulated_row(): r = _rec(quantity=20, entry_price=9000.0) # 10@10000 + 10@8000(주문가) dbw._apply_entry_price_adjustments_v2( r, {"entry_fill_adjust": {"order_price": 8000.0, "filled_price": 7900.0, "filled_qty": 10}}) assert r.entry_price == 8950.0 def test_fill_adjust_bad_input_keeps_price(): r = _rec() dbw._apply_entry_price_adjustments_v2( r, {"entry_fill_adjust": {"order_price": 0, "filled_price": 7900.0, "filled_qty": 10}}) assert r.entry_price == 10000.0 def test_upsert_handler_skips_fill_adjust_when_no_row(): class _S: async def execute(self, q): return SimpleNamespace(scalars=lambda: SimpleNamespace(first=lambda: None)) def add(self, x): # pragma: no cover raise AssertionError("must not create") async def commit(self): # pragma: no cover raise AssertionError("must not commit") _run(dbw._handle_upsert_active_trade(_S(), { "stock_code": "005930", "strategy_name": "swing", "entry_fill_adjust": {"order_price": 1, "filled_price": 1, "filled_qty": 1}, })) # ── 2. 스윙 진입가 해석 (손절 판정 직결 — 경계값) ──────────────────────────── def test_entry_price_prefers_active_trade_within_band(): assert resolve_swing_entry_price_v2(68000, 67323.08) == (68000.0, "active_trade") def test_entry_price_boundary_exact_deviation_uses_active_trade(): pa = 10000.0 at = pa * (1 + SWING_ENTRY_PRICE_MAX_DEVIATION) price, basis = resolve_swing_entry_price_v2(at, pa) assert basis == "active_trade" and price == pytest.approx(at) def test_entry_price_over_deviation_uses_kis_portfolio_avg(): # [2026-10-09 하드닝 R1] 괴리 > 한도면 유령 행 의심 → KIS 실제 평단(Portfolio). # 이전 정책(max)은 유령 고가 entry 로 허위 전량 손절을 냈다. pa = 10000.0 lo = pa * (1 - SWING_ENTRY_PRICE_MAX_DEVIATION - 0.01) assert resolve_swing_entry_price_v2(lo, pa) == (pa, "portfolio_avg_deviation") hi = pa * (1 + SWING_ENTRY_PRICE_MAX_DEVIATION + 0.01) assert resolve_swing_entry_price_v2(hi, pa) == (pa, "portfolio_avg_deviation") @pytest.mark.parametrize("at,pa,exp", [ (None, 5000, (5000.0, "portfolio_avg")), (0, 5000, (5000.0, "portfolio_avg")), ("bad", 5000, (5000.0, "portfolio_avg")), (5000, 0, (5000.0, "active_trade")), (None, None, (0.0, "none")), ]) def test_entry_price_fallbacks(at, pa, exp): assert resolve_swing_entry_price_v2(at, pa) == exp def test_executor_entry_price_dt_tag_uses_portfolio(): h = SimpleNamespace(average_price=Decimal("67323.08")) assert te._swing_entry_price_for_stock_v2( {"strategy_type": "daytrading", "_swing_entry_price": 68000}, h, "103140") == 67323.08 assert te._swing_entry_price_for_stock_v2( {"_swing_entry_price": 68000}, h, "103140") == 68000.0 assert te._swing_entry_price_for_stock_v2({}, h, "103140") == 67323.08 def test_mixed_avg_stop_loss_trigger_differs(): """혼합 평단(67,323)으로는 -7% 미발동인데 스윙 진입가(68,000)로는 발동 — 스윙 기준이 맞다.""" stop = -abs(float(settings.STOP_LOSS_PERCENT)) price = 68000 * (1 + stop / 100) - 1 swing_pl = (price / 68000 - 1) * 100 mixed_pl = (price / 67323.08 - 1) * 100 assert swing_pl <= stop < mixed_pl # ── 2-b. API 전략별 분해 필드 ──────────────────────────────────────────────── def _h(**kw): base = dict(stock_code="103140", quantity=13, average_price=Decimal("67323.08"), current_price=Decimal("70000")) base.update(kw) return SimpleNamespace(**base) def test_breakdown_swing_with_dt(): b = pf._strategy_breakdown_v2(_h(), 68000.0, {"quantity": 8, "entry_price": 66900.0}) assert b["swing_quantity"] == 5 and b["dt_quantity"] == 8 assert b["swing_avg_price"] == 68000.0 and b["swing_avg_price_basis"] == "active_trade" assert b["dt_avg_price"] == 66900.0 assert b["swing_profit_loss"] == pytest.approx(10000.0) assert b["avg_price_mixed"] is True def test_breakdown_derived_when_no_active_trade(): b = pf._strategy_breakdown_v2(_h(), None, {"quantity": 8, "entry_price": 66900.0}) assert b["swing_avg_price_basis"] == "derived_from_mixed" assert b["swing_avg_price"] == pytest.approx(round((67323.08 * 13 - 66900 * 8) / 5, 2)) def test_breakdown_no_dt_portfolio_fallback_rounded(): b = pf._strategy_breakdown_v2(_h(quantity=5, average_price=Decimal("67323.0769")), None, None) assert b["swing_avg_price"] == 67323.08 and b["swing_avg_price_basis"] == "portfolio_avg" assert b["dt_quantity"] == 0 and b["avg_price_mixed"] is False def test_breakdown_dt_only(): b = pf._strategy_breakdown_v2(_h(quantity=8), None, {"quantity": 8, "entry_price": 66900.0}) assert b["swing_quantity"] == 0 and b["swing_avg_price"] is None def test_quantize_price_policy(): assert pf._quantize_price_v2("67323.0769") == Decimal("67323.08") assert pf._quantize_price_v2("1.005") == Decimal("1.01") assert pf._quantize_price_v2(None) == Decimal("0.00") def test_response_model_has_swing_avg_price(): assert "swing_avg_price" in pf.PortfolioResponse.model_fields # ── 3. 환율 ────────────────────────────────────────────────────────────────── @pytest.fixture def fx_reset(monkeypatch): monkeypatch.setattr(pf, "_last_good_fx_rate", 0.0) yield def test_fx_from_kis_and_last_good(fx_reset, monkeypatch): assert pf._resolve_fx_rate_v2({"output2": [{"bass_exrt": "1395.5"}]}) == (1395.5, "kis") assert pf._resolve_fx_rate_v2({"output2": []}) == (1395.5, "last_good") def test_fx_out_of_range_ignored_then_config(fx_reset): rate, src = pf._resolve_fx_rate_v2({"output2": {"bass_exrt": "3.2"}}) assert src == "config_default" and rate == float(settings.US_DEFAULT_FX_RATE) def test_fx_unavailable_when_all_zero(fx_reset, monkeypatch): monkeypatch.setattr(settings, "US_DEFAULT_FX_RATE", 0.0) assert pf._resolve_fx_rate_v2(None) == (0.0, "unavailable") def test_fx_derived_ratio_no_double_count(fx_reset): data = {"output2": {"tot_evlu_amt": "1400000", "frcr_evlu_tota": "1000", "usd_amt": "1000"}} assert pf._resolve_fx_rate_v2(data) == (1400.0, "kis") def test_cache_summary_v2_fx_and_cash_fields(fx_reset, monkeypatch): cache = pf.PortfolioCache() cache.holdings = [ SimpleNamespace(stock_code="005930", market="KRX", quantity=10, average_price=Decimal("70000"), current_price=Decimal("71000"), strategy_type="swing", profit_loss_rate=None), SimpleNamespace(stock_code="AAPL", market="NASD", quantity=2, average_price=Decimal("150"), current_price=Decimal("200"), strategy_type="swing", profit_loss_rate=None), ] bal = {"output2": [{"tot_evlu_amt": "1000000", "scts_evlu_amt": "710000", "prvs_rcdl_excc_amt": "290000", "_ord_psbl_cash": "264854", "dnca_tot_amt": "290000"}]} _run(cache.cache_summary_v2(bal, {"output2": []})) s = cache.summary # [R1 2026-10-09] 설정 기본값 환율은 추정치 — 총자산·원화 환산에 합산하지 않고 별도 노출 assert s["exchange_rate"] == 0.0 assert s["exchange_rate_estimate"] == float(settings.US_DEFAULT_FX_RATE) assert s["exchange_rate_source"] == "config_default" and s["exchange_rate_unavailable"] is True assert s["overseas_value_usd"] == 400.0 assert s["overseas_value_krw"] == 0.0 assert s["available_cash"] == 264854.0 # 주문가능현금(매매 사이징 입력) — 의미 불변 assert s["settlement_cash"] == 290000.0 assert s["cash_basis_gap"] == pytest.approx(25146.0) assert s["cash_for_total_assets"] == 290000.0 assert s["total_assets_residual"] == 0.0 def test_cache_summary_v2_fx_unavailable_flag(fx_reset, monkeypatch): monkeypatch.setattr(settings, "US_DEFAULT_FX_RATE", 0.0) cache = pf.PortfolioCache() cache.holdings = [SimpleNamespace(stock_code="AAPL", market="NASD", quantity=2, average_price=Decimal("150"), current_price=Decimal("200"), strategy_type="swing", profit_loss_rate=None)] _run(cache.cache_summary_v2(None, None)) s = cache.summary assert s["exchange_rate"] == 0.0 and s["exchange_rate_unavailable"] is True assert s["fx_conversion_skipped"] is True and s["overseas_value_krw"] == 0.0 # ── 1-c. 후속 체결 보정(fill_sweep) ────────────────────────────────────────── def test_derive_sell_entry_price(): row = SimpleNamespace(order_quantity=10, total_amount=Decimal("110000"), net_profit_loss=Decimal("9790"), commission=Decimal("210")) assert fs.derive_sell_entry_price_v2(row) == pytest.approx(10000.0) row2 = SimpleNamespace(order_quantity=10, total_amount=110000, net_profit_loss=None, commission=None) assert fs.derive_sell_entry_price_v2(row2) is None class _FakeDB: """1번째 execute = TradeHistory 후보, 이후 = ActiveTrade 조회(진입가 보정 대상 확정).""" def __init__(self, rows, active_rows=None): self.rows = rows self.active_rows = list(active_rows or []) self.calls = 0 async def execute(self, q): self.calls += 1 rows = self.rows if self.calls == 1 else self.active_rows return SimpleNamespace(scalars=lambda: SimpleNamespace(all=lambda: rows)) def _th_row(**kw): base = dict(id=448, stock_code="082740", order_type="BUY", order_id="0000777", order_quantity=9, order_price=Decimal("41506"), total_amount=Decimal("373554"), net_profit_loss=None, commission=None, strategy_type="swing", entry_reason="RSI 과매도 반등") base.update(kw) return SimpleNamespace(**base) @pytest.fixture def sweep_env(monkeypatch): calls = {"th": [], "at": []} async def _th(tid, fields): calls["th"].append((tid, dict(fields))) async def _at(data): calls["at"].append(dict(data)) monkeypatch.setattr(dbw, "enqueue_update_trade_history", _th) monkeypatch.setattr(dbw, "enqueue_active_trade_upsert", _at) fs._attempts.clear() fs._queued_ids.clear() te._entry_fill_adjusted_orders.clear() return calls def test_sweep_buy_updates_price_amount_and_entry(monkeypatch, sweep_env): _patch_exec(monkeypatch, {"ok": True, "found": True, "filled_qty": 9, "filled_price": 41200.0, "remaining_qty": 0}) active = [SimpleNamespace(id=225, strategy_name="swing", entry_reason="RSI 과매도 반등")] n = _run(fs.sweep_unconfirmed_fills_v2(_FakeDB([_th_row()], active))) assert n == 1 (tid, f), = sweep_env["th"] assert tid == 448 and f["executed_price"] == Decimal("41200.0") assert f["total_amount"] == Decimal(str(41200.0 * 9)) assert "executed_at" not in f (at,) = sweep_env["at"] # [하드닝 R1 2026-10-09] '최신 비-DT 행'이 아니라 이 매수의 전략명으로 정확히 지정 assert at["strategy_name"] == "swing" and "strategy_name_not" not in at assert at["entry_fill_adjust"]["filled_price"] == 41200.0 def test_sweep_buy_skips_entry_adjust_if_already_done(monkeypatch, sweep_env): te.note_entry_fill_adjusted_v2("777") _patch_exec(monkeypatch, {"ok": True, "found": True, "filled_qty": 9, "filled_price": 41200.0, "remaining_qty": 0}) _run(fs.sweep_unconfirmed_fills_v2(_FakeDB([_th_row()]))) assert len(sweep_env["th"]) == 1 and sweep_env["at"] == [] def test_sweep_no_fill_price_no_update_and_attempts_capped(monkeypatch, sweep_env): _patch_exec(monkeypatch, {"ok": True, "found": True, "filled_qty": 9, "filled_price": 0, "remaining_qty": 0}) monkeypatch.setattr(fs, "FILL_SWEEP_RETRY_INTERVAL_SEC", 0) for _ in range(fs.FILL_SWEEP_MAX_ATTEMPTS + 2): _run(fs.sweep_unconfirmed_fills_v2(_FakeDB([_th_row()]))) assert sweep_env["th"] == [] assert fs._attempts[448][0] == fs.FILL_SWEEP_MAX_ATTEMPTS assert 448 in fs._exhausted_ids() def test_sweep_sell_recomputes_net(monkeypatch, sweep_env): row = _th_row(id=500, order_type="SELL", order_quantity=10, order_price=Decimal("11000"), total_amount=Decimal("110000"), net_profit_loss=Decimal("10000"), commission=None) _patch_exec(monkeypatch, {"ok": True, "found": True, "filled_qty": 10, "filled_price": 10900.0, "remaining_qty": 0}) _run(fs.sweep_unconfirmed_fills_v2(_FakeDB([row]))) (tid, f), = sweep_env["th"] assert f["total_amount"] == Decimal("109000.0") assert float(f["net_profit_loss"]) < 9000.0 # (10900-10000)×10 − 비용 assert float(f["profit_loss_rate"]) == pytest.approx(9.0) assert sweep_env["at"] == [] def test_sweep_partial_unsettled_skipped(monkeypatch, sweep_env): _patch_exec(monkeypatch, {"ok": True, "found": True, "filled_qty": 4, "filled_price": 41200.0, "remaining_qty": 5}) _run(fs.sweep_unconfirmed_fills_v2(_FakeDB([_th_row()]))) assert sweep_env["th"] == [] def test_sweep_skips_overseas(monkeypatch, sweep_env): m = _patch_exec(monkeypatch, {"ok": True}) _run(fs.sweep_unconfirmed_fills_v2(_FakeDB([_th_row(stock_code="AAPL")]))) m.assert_not_called()