From 8ecedf65a8ee548f15922f53c11fc27dcd87d248 Mon Sep 17 00:00:00 2001 From: shuim Date: Tue, 22 Sep 2026 10:32:14 +0900 Subject: [PATCH] =?UTF-8?q?fix:=20=EC=8A=A4=EC=9C=99=20=EB=B6=80=EB=B6=84?= =?UTF-8?q?=EB=A7=A4=EB=8F=84=20=EC=A4=91=EB=B3=B5=20=EC=8B=A4=ED=96=89=20?= =?UTF-8?q?=E2=80=94=20ActiveTrade=20=EB=AF=B8=EC=A1=B4=EC=9E=AC=20?= =?UTF-8?q?=ED=8F=AC=EC=A7=80=EC=85=98=20=EC=B2=98=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 동기화로 유입된 포지션(ActiveTrade 행 없음)에 1차 익절(50%)이 매 사이클 반복 실행돼 전량 청산되는 문제 수정 (2026-09-22 두산에너빌리티 18→9→0주). - exit_service: 부분매도 시 ActiveTrade 없으면 필수 키 포함해 생성, Portfolio 잔량 즉시 감소 - db_worker: 필수 키 없는 ActiveTrade 신규 upsert는 재시도 대신 경고 후 스킵 - portfolio.py sync 청산 경로 동일 수정 - 회귀 테스트 추가 Co-Authored-By: Claude Opus 5 (1M context) --- backend/app/api/endpoints/portfolio.py | 24 +++- backend/app/services/db_worker.py | 9 ++ backend/app/services/swing/exit_service.py | 28 +++- ...test_swing_partial_exit_no_active_trade.py | 123 ++++++++++++++++++ 4 files changed, 180 insertions(+), 4 deletions(-) create mode 100644 backend/tests/test_swing_partial_exit_no_active_trade.py diff --git a/backend/app/api/endpoints/portfolio.py b/backend/app/api/endpoints/portfolio.py index ca469c2..8644c22 100644 --- a/backend/app/api/endpoints/portfolio.py +++ b/backend/app/api/endpoints/portfolio.py @@ -1139,9 +1139,29 @@ async def check_and_execute_exit_trades(db: AsyncSession): # 분할 청산 시 ActiveTrade 잔여 비율 업데이트 # TODO: db_worker 마이그레이션 필요 — 읽기+업데이트가 같은 트랜잭션이라 큐 분리 어려움 - if exit_signal and exit_signal.action == "SELL_PARTIAL" and active_trade: + if exit_signal and exit_signal.action == "SELL_PARTIAL" and sell_quantity < quantity: new_remaining = remaining_pct * (1.0 - exit_signal.sell_pct) - active_trade.remaining_pct = Decimal(str(round(new_remaining, 4))) + # 잔여 수량 즉시 반영 (다음 사이클 원래 수량 기준 재매도 방지) + holding.quantity = quantity - sell_quantity + holding.current_price = Decimal(str(latest_price)) + if active_trade: + active_trade.remaining_pct = Decimal(str(round(new_remaining, 4))) + else: + # [FIX] ActiveTrade 없는 포지션(동기화 유입)은 remaining_pct를 + # 저장할 곳이 없어 부분매도가 매 사이클 반복됨 → 행 생성 + db.add(ActiveTrade( + stock_code=stock_code, + stock_name=stock_name, + quantity=quantity - sell_quantity, + entry_price=Decimal(str(avg_price)), + current_price=Decimal(str(latest_price)), + status="active", + remaining_pct=Decimal(str(round(new_remaining, 4))), + highest_since_entry=Decimal(str(latest_price)), + strategy_name="swing", + entry_reason="동기화 유입 포지션 (부분매도 시 자동 생성)", + currency_code="KRW" if domestic else "USD", + )) await db.commit() else: # 전량 매도 diff --git a/backend/app/services/db_worker.py b/backend/app/services/db_worker.py index 925dc87..5c0de52 100644 --- a/backend/app/services/db_worker.py +++ b/backend/app/services/db_worker.py @@ -552,6 +552,15 @@ async def db_writer_worker() -> None: if record is None: # remaining_pct_multiply는 신규 생성 시 무시 create_data = {k: v for k, v in data.items() if k != "remaining_pct_multiply"} + # 필수 컬럼(quantity/entry_price) 없이 신규 생성 요청이 오면 + # IntegrityError로 3회 재시도만 낭비 → 경고 후 스킵 + _missing = [k for k in ("quantity", "entry_price") if create_data.get(k) is None] + if _missing: + logger.warning( + "⚠️ [Worker] ActiveTrade upsert 스킵: %s active 행 없음 + 필수 키 누락 %s (keys=%s)", + stock_code, _missing, sorted(create_data.keys()), + ) + continue record = ActiveTrade(**create_data) session.add(record) else: diff --git a/backend/app/services/swing/exit_service.py b/backend/app/services/swing/exit_service.py index 7c2ab97..2ee49d8 100644 --- a/backend/app/services/swing/exit_service.py +++ b/backend/app/services/swing/exit_service.py @@ -341,11 +341,35 @@ async def check_and_execute_exit_v2(db: AsyncSession) -> list[dict]: else: # 분할 매도 후 remaining_pct 갱신 (반복 트리거 방지) new_remaining = max(0.0, remaining_pct - sell_pct) - from app.services.db_worker import enqueue_active_trade_upsert - await enqueue_active_trade_upsert({ + from app.services.db_worker import ( + enqueue_active_trade_upsert, enqueue_portfolio_upsert, + ) + _at_data = { "stock_code": stock_code, "remaining_pct": new_remaining, "highest_since_entry": latest_price if latest_price > (trailing_high or 0) else trailing_high, + } + if active_trade is None: + # [FIX] ActiveTrade 행이 없는 포지션(동기화 유입 등)은 upsert가 + # NOT NULL(quantity/entry_price) 위반으로 INSERT 실패 → remaining_pct + # 미저장 → 다음 사이클에 50% 매도 재실행(2026-09-22 두산에너빌리티). + # 신규 생성에 필요한 필수 키를 함께 전달한다. + _at_data.update({ + "stock_name": stock_name, + "quantity": quantity - sell_qty, + "entry_price": avg_price, + "current_price": latest_price, + "strategy_name": "swing", + "status": "active", + "entry_reason": "동기화 유입 포지션 (부분매도 시 자동 생성)", + }) + await enqueue_active_trade_upsert(_at_data) + # [FIX] 잔여 수량도 즉시 반영 — 다음 사이클이 동기화 전 원래 수량 + # 기준으로 매도 수량을 계산하지 않도록 (18주→9주 재매도 방지) + await enqueue_portfolio_upsert({ + "stock_code": stock_code, + "quantity": quantity - sell_qty, + "current_price": latest_price, }) # LossTracker 업데이트 — 전량매도 시에만 (분할매도 이중 카운트 방지) diff --git a/backend/tests/test_swing_partial_exit_no_active_trade.py b/backend/tests/test_swing_partial_exit_no_active_trade.py new file mode 100644 index 0000000..5b2b903 --- /dev/null +++ b/backend/tests/test_swing_partial_exit_no_active_trade.py @@ -0,0 +1,123 @@ +"""ActiveTrade 행이 없는 스윙 포지션의 부분매도 중복 실행 회귀 테스트. + +2026-09-22 두산에너빌리티(034020): 동기화로 유입된 포지션(ActiveTrade 없음)에 +"1차 익절(50%)"이 2분 간격으로 두 번 실행돼 전량 청산됨. +원인: exit_service가 remaining_pct 갱신 upsert에 quantity/entry_price를 넣지 않아 +db_worker의 신규 INSERT가 NOT NULL 위반 → remaining_pct 미저장 + Portfolio.quantity 미감소. +""" +import asyncio +from decimal import Decimal +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine +from sqlalchemy import select + +from app.models.models import Base, Portfolio, ActiveTrade +from app.services.mean_reversion_strategy import ExitSignal +from app.services import db_worker +from app.services.swing import exit_service + + +def _daily_chart(n=30, close=86200.0): + return {"output2": [ + {"stck_clpr": str(close), "stck_hgpr": str(close * 1.01), "stck_lwpr": str(close * 0.99)} + for _ in range(n) + ]} + + +async def _drain(queue, session_factory): + """db_worker 로직으로 큐를 처리 (실제 워커 코드 경로 사용).""" + with patch.object(db_worker, "AsyncSessionLocal", session_factory), \ + patch.object(db_worker, "get_db_task_queue", return_value=queue): + worker = asyncio.create_task(db_worker.db_writer_worker()) + await queue.join() + worker.cancel() + try: + await worker + except (asyncio.CancelledError, Exception): + pass + + +@pytest.mark.asyncio +async def test_partial_exit_without_active_trade_runs_once(): + 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=[Portfolio.__table__, ActiveTrade.__table__])) + Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + async with Session() as s: + s.add(Portfolio( + stock_code="034020", stock_name="두산에너빌리티", quantity=18, + average_price=Decimal("85589"), current_price=Decimal("86200"), + strategy_type="swing", + )) + await s.commit() + + queue = asyncio.Queue() + partial = ExitSignal(action="SELL_PARTIAL", reason="1차 익절(50%): RSI(55.2)>=55", sell_pct=0.5) + order_ok = {"rt_cd": "0", "output": {"ODNO": "0000002153"}} + + api = SimpleNamespace( + get_current_price=AsyncMock(return_value={"output": {"stck_prpr": "86200"}}), + get_daily_chart=AsyncMock(return_value=_daily_chart()), + ) + place_order = AsyncMock(return_value=order_ok) + + with patch.object(exit_service, "korea_investment_api", api), \ + patch.object(exit_service, "check_market_regime", AsyncMock(return_value={"regime_200d": "bull"})), \ + patch.object(exit_service, "check_exit_signal", return_value=partial), \ + patch.object(exit_service, "calculate_atr", return_value=1500.0), \ + patch.object(exit_service, "calculate_rsi", return_value=55.2), \ + patch.object(exit_service, "calculate_bollinger", return_value=(90000.0, 85370.0, 80000.0)), \ + patch("app.services.trade_executor.safe_place_order", place_order), \ + patch.object(db_worker, "db_task_queue", queue), \ + patch.object(exit_service.loss_tracker, "record_trade_result_safe", AsyncMock()), \ + patch.object(exit_service.loss_tracker, "save_to_db", AsyncMock()): + + # 1차 사이클: 18주 → 9주 매도 + async with Session() as s: + exits = await exit_service.check_and_execute_exit_v2(s) + assert len(exits) == 1 and exits[0]["sell_quantity"] == 9 + await _drain(queue, Session) + + async with Session() as s: + at = (await s.execute(select(ActiveTrade).where(ActiveTrade.stock_code == "034020"))).scalars().all() + pf = (await s.execute(select(Portfolio).where(Portfolio.stock_code == "034020"))).scalar_one() + assert len(at) == 1, "부분매도 후 ActiveTrade 행이 생성돼야 함" + assert float(at[0].remaining_pct) == 0.5 + assert at[0].quantity == 9 and float(at[0].entry_price) == 85589.0 + assert pf.quantity == 9, "부분매도 후 Portfolio 잔량이 즉시 감소해야 함" + + # 2차 사이클: remaining_pct=0.5 → check_exit_signal에 전달되는 값으로 재실행 차단 확인 + seen = {} + def _capture(**kw): + seen.update(kw) + return None + with patch.object(exit_service, "check_exit_signal", side_effect=_capture): + async with Session() as s: + exits2 = await exit_service.check_and_execute_exit_v2(s) + assert exits2 == [] + assert seen["remaining_pct"] == 0.5 + assert place_order.await_count == 1, "매도 주문은 한 번만 나가야 함" + + await engine.dispose() + + +@pytest.mark.asyncio +async def test_db_worker_skips_active_trade_insert_without_required_keys(): + """필수 키 없는 신규 upsert는 IntegrityError 재시도 대신 스킵.""" + 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=[Portfolio.__table__, ActiveTrade.__table__])) + Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + queue = asyncio.Queue() + await queue.put({"type": "upsert_active_trade", "data": {"stock_code": "000000", "remaining_pct": 0.5}}) + await asyncio.wait_for(_drain(queue, Session), timeout=5) # 재시도(1+2+4초) 없이 즉시 끝나야 함 + + async with Session() as s: + rows = (await s.execute(select(ActiveTrade))).scalars().all() + assert rows == [] + await engine.dispose()