Merge branch 'dev' into main — 스윙 부분매도 중복 실행 수정

동기화로 유입된 포지션(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) <noreply@anthropic.com>

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)
+                            # 잔여 수량 즉시 반영 (다음 사이클 원래 수량 기준 재매도 방지)
+                            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:
                             # 전량 매도

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:

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 업데이트 — 전량매도 시에만 (분할매도 이중 카운트 방지)

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()