shuim
a264561576
- 스윙 체결가 확인 전 executed_price NULL + fill_sweep, 진입가 resolve(ActiveTrade↔Portfolio), 0체결/부분체결 매수 장부 원복 - DT 고아 감지(근거 한정)·0체결 매도 복원·stale 점검 확대, 시간손절 -0.3% 원복 - 폴백 손절 ATR 캐시, 거래량부족-차단 신호 매수 차단, 현금 5% 유보 재적용, MDD peak/reset 수정 - KIS 서킷브레이커 청산 우회·경보, db_worker 재큐잉 순서, heartbeat·health-watch 사이드카 - 실시간 WS 재구독·41건 우선순위·다건 프레임 분할, 차트 샘플 제거·당일봉 합성, WS 서브프로토콜 인증 - 프론트: 미관리 그룹·확인 모달·차트 실시간 반영·예수금 표시 정정 - 전문가 에이전트 6종 추가(.claude/agents) - 유닛 스위트 2667 passed (소켓 차단 하), tsc OK Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y83Za98XsNw9zuA4FTUytN
387 lines
16 KiB
Python
387 lines
16 KiB
Python
"""스윙 포지션 관리 -- DB 조회 전용
|
||
|
||
strategy_type='swing' 필터를 적용해 데이트레이딩 포지션과 완전히 분리.
|
||
모든 함수는 읽기 전용 -- 쓰기는 db_worker.py의 큐를 통해서만 수행.
|
||
|
||
DB 테이블 매핑:
|
||
- Portfolio: 보유 종목 (strategy_type 컬럼으로 swing/daytrading 구분)
|
||
- ActiveTrade: 활성 매매 (strategy_name 컬럼으로 구분, 분할 청산 상태 추적)
|
||
|
||
사용 위치:
|
||
- exit_service.py: 청산 대상 포지션 조회
|
||
- entry_service.py: 포지션 한도 체크
|
||
- scheduler.py: 스윙 스케줄러 사이클
|
||
"""
|
||
import logging
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.models.models import Portfolio, ActiveTrade
|
||
# [§8.5 항목3] DT 제외 필터는 db_worker 를 정본으로 공유한다.
|
||
# (db_worker 는 서비스 모듈을 모듈 수준에서 import 하지 않으므로 순환 참조 없음)
|
||
from app.services.db_worker import DT_STRATEGY_NAMES, not_dt_strategy_clause # noqa: F401
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 스윙 전략 식별자 -- Portfolio.strategy_type과 매칭
|
||
STRATEGY_TYPE = "swing"
|
||
|
||
|
||
async def get_swing_holdings(db: AsyncSession) -> list:
|
||
"""strategy_type='swing'인 보유 종목 목록 반환.
|
||
|
||
Portfolio 테이블에서 수량 > 0이고 strategy_type='swing'인 레코드 조회.
|
||
|
||
Args:
|
||
db: AsyncSession 인스턴스
|
||
|
||
Returns:
|
||
list[Portfolio]: 스윙 보유 종목 ORM 객체 목록
|
||
"""
|
||
result = await db.execute(
|
||
select(Portfolio).where(
|
||
Portfolio.quantity > 0,
|
||
Portfolio.strategy_type == STRATEGY_TYPE,
|
||
)
|
||
)
|
||
holdings = result.scalars().all()
|
||
# [2026-10-09 하드닝 R1] 어느 청산 엔진에도 속하지 않는 보유분 탐지(경보 전용).
|
||
# 경보는 리스크 가드가 아니므로 실패해도 청산 대상 반환을 막지 않는다(청산 우선 §8.1-11).
|
||
try:
|
||
await detect_strategy_tag_mismatch_v2(db)
|
||
except Exception as e:
|
||
logger.warning(f"[SwingPM] 전략 태그 불일치 탐지 실패(청산 대상 조회는 계속): {e}")
|
||
return holdings
|
||
|
||
|
||
# ── [2026-10-09 하드닝 R1] Portfolio 태그 ↔ ActiveTrade 전략명 불일치 탐지 ─────────
|
||
# 사례: 105560 3주가 Portfolio.strategy_type='daytrading' + active ActiveTrade
|
||
# strategy_name='synced'(비-DT). get_swing_holdings 는 'swing' 태그만 보므로 스윙 청산이
|
||
# 보지 않고, DT 엔진은 인메모리 포지션만 관리하므로(당일 매수분은 EOD 청산) 잔여 수량에는
|
||
# **손절이 걸려 있지 않다**. 자동 편입(스윙 청산 대상으로 승격)은 과다매도·소유권 규칙
|
||
# (trade_executor._sell_ownership_block_reason_v2) 충돌 위험이 있어 하지 않고, error 경보로
|
||
# 운영자가 KIS 실보유를 확인하게 한다.
|
||
def find_strategy_tag_mismatches_v2(
|
||
portfolio_rows,
|
||
active_rows,
|
||
dt_inmemory_qty: dict | None = None,
|
||
) -> list[dict]:
|
||
"""순수 함수 — DT 태그 Portfolio 행 중 비-DT active ActiveTrade 가 걸려 있고,
|
||
DT 인메모리 수량이 보유 수량을 덮지 못하는(= 관리 주체 없는 수량이 있는) 종목.
|
||
|
||
Args:
|
||
portfolio_rows: Portfolio 유사 객체(stock_code, quantity, strategy_type)
|
||
active_rows: status='active' ActiveTrade 유사 객체(stock_code, strategy_name, id)
|
||
dt_inmemory_qty: {stock_code: DT 인메모리 보유 수량}
|
||
|
||
Returns:
|
||
[{stock_code, portfolio_qty, portfolio_tag, dt_inmemory_qty, unmanaged_qty,
|
||
active_trade_ids, active_strategy_names}]
|
||
"""
|
||
dt_mem = dt_inmemory_qty or {}
|
||
non_dt_active: dict[str, list] = {}
|
||
for a in active_rows or []:
|
||
name = getattr(a, "strategy_name", None)
|
||
if name in DT_STRATEGY_NAMES:
|
||
continue
|
||
non_dt_active.setdefault(getattr(a, "stock_code", None), []).append(a)
|
||
|
||
out: list[dict] = []
|
||
for p in portfolio_rows or []:
|
||
if getattr(p, "strategy_type", None) not in DT_STRATEGY_NAMES:
|
||
continue
|
||
try:
|
||
qty = int(getattr(p, "quantity", 0) or 0)
|
||
except (TypeError, ValueError):
|
||
qty = 0
|
||
if qty <= 0:
|
||
continue
|
||
code = getattr(p, "stock_code", None)
|
||
rows = non_dt_active.get(code)
|
||
if not rows:
|
||
continue
|
||
try:
|
||
mem = max(0, int(dt_mem.get(code, 0) or 0))
|
||
except (TypeError, ValueError):
|
||
mem = 0
|
||
unmanaged = qty - min(mem, qty)
|
||
if unmanaged <= 0:
|
||
continue
|
||
out.append({
|
||
"stock_code": code,
|
||
"portfolio_qty": qty,
|
||
"portfolio_tag": getattr(p, "strategy_type", None),
|
||
"dt_inmemory_qty": mem,
|
||
"unmanaged_qty": unmanaged,
|
||
"active_trade_ids": [getattr(r, "id", None) for r in rows],
|
||
"active_strategy_names": [getattr(r, "strategy_name", None) for r in rows],
|
||
})
|
||
return out
|
||
|
||
|
||
def _dt_inmemory_qty_snapshot_v2() -> dict:
|
||
"""DT 인메모리 보유 수량 스냅샷(읽기 전용). DT 상태를 못 읽으면 빈 dict —
|
||
그 경우 DT 태그 수량 전부를 '관리 주체 없음'으로 보아 경보 쪽으로 기운다."""
|
||
try:
|
||
from app.services.daytrading.position_manager import day_trading_state
|
||
return {
|
||
code: int((pos or {}).get("quantity", 0) or 0)
|
||
for code, pos in dict(day_trading_state.positions).items()
|
||
}
|
||
except Exception as e:
|
||
logger.warning(f"[SwingPM] DT 인메모리 수량 조회 실패 — 빈 스냅샷으로 판정: {e}")
|
||
return {}
|
||
|
||
|
||
async def detect_strategy_tag_mismatch_v2(db: AsyncSession) -> list[dict]:
|
||
"""DB SELECT 2회(쓰기 없음) + DT 인메모리 대조 → 불일치 종목을 error 로그(종목별 레이트리밋).
|
||
|
||
반환값은 테스트·상태 조회용. 자동 수정은 하지 않는다(운영자 확인 필요).
|
||
"""
|
||
p_res = await db.execute(
|
||
select(Portfolio).where(
|
||
Portfolio.quantity > 0,
|
||
Portfolio.strategy_type.in_(DT_STRATEGY_NAMES),
|
||
)
|
||
)
|
||
p_rows = p_res.scalars().all()
|
||
if not p_rows:
|
||
return []
|
||
codes = [getattr(p, "stock_code", None) for p in p_rows]
|
||
a_res = await db.execute(
|
||
select(ActiveTrade).where(
|
||
ActiveTrade.status == "active",
|
||
ActiveTrade.stock_code.in_(codes),
|
||
not_dt_strategy_clause(ActiveTrade.strategy_name),
|
||
)
|
||
)
|
||
a_rows = a_res.scalars().all()
|
||
found = find_strategy_tag_mismatches_v2(p_rows, a_rows, _dt_inmemory_qty_snapshot_v2())
|
||
if found:
|
||
from app.core.log_throttle import should_log
|
||
for m in found:
|
||
ok, sup = should_log(f"swing_pm:tag_mismatch:{m['stock_code']}")
|
||
if not ok:
|
||
continue
|
||
logger.error(
|
||
f"[SwingPM] 🚨 {m['stock_code']}: Portfolio 태그 '{m['portfolio_tag']}' "
|
||
f"{m['portfolio_qty']}주인데 비-DT active ActiveTrade "
|
||
f"{m['active_strategy_names']}(id={m['active_trade_ids']}) 존재, DT 인메모리 "
|
||
f"{m['dt_inmemory_qty']}주 → {m['unmanaged_qty']}주가 스윙·DT 어느 청산 엔진에도 "
|
||
f"속하지 않아 손절 없음. KIS 실보유 확인 후 태그/ActiveTrade 수동 정리 필요"
|
||
+ (f" (+{sup}건 억제)" if sup else "")
|
||
)
|
||
return found
|
||
|
||
|
||
async def get_swing_active_trades(db: AsyncSession) -> list:
|
||
"""활성 스윙 ActiveTrade 목록 반환.
|
||
|
||
DT 계열(DT_STRATEGY_NAMES) 제외 조건으로 데이트레이딩 포지션만 걸러낸다.
|
||
분할 청산 진행 상태(remaining_pct), 최고가(highest_since_entry) 포함.
|
||
|
||
[2026-09-27 §8.5 항목3] 기존 `strategy_name != "day_trade"` 는
|
||
SQL 3값 논리로 strategy_name IS NULL 인 legacy 스윙 행까지 함께 제외했고,
|
||
'daytrading' 전략명은 반대로 통과시켰다 → not_dt_strategy_clause() 로 통일.
|
||
|
||
Args:
|
||
db: AsyncSession 인스턴스
|
||
|
||
Returns:
|
||
list[ActiveTrade]: 활성 스윙 매매 ORM 객체 목록
|
||
"""
|
||
result = await db.execute(
|
||
select(ActiveTrade).where(
|
||
ActiveTrade.status == "active",
|
||
not_dt_strategy_clause(ActiveTrade.strategy_name),
|
||
)
|
||
)
|
||
return result.scalars().all()
|
||
|
||
|
||
async def get_swing_active_trade(db: AsyncSession, stock_code: str) -> ActiveTrade | None:
|
||
"""특정 종목의 스윙 ActiveTrade 반환.
|
||
|
||
분할 청산 상태(remaining_pct)와 최고가(highest_since_entry) 조회용.
|
||
청산 로직에서 전략별 분기(모멘텀/평균회귀)에도 사용.
|
||
|
||
Args:
|
||
db: AsyncSession 인스턴스
|
||
stock_code: 종목 코드
|
||
|
||
Returns:
|
||
ActiveTrade | None: 활성 매매 ORM 객체, 없으면 None
|
||
"""
|
||
# [2026-09-23] scalar_one_or_none() 은 행이 2개 이상이면 MultipleResultsFound 를 던지고,
|
||
# 호출측 except 가 이를 삼켜 해당 종목의 청산(손절 포함)이 통째로 스킵됐다.
|
||
# db_worker legacy 경로와 동일하게 가장 최근 행을 사용한다.
|
||
# [2026-09-27 §8.5 항목3] strategy_name IS NULL 인 legacy 스윙 행이
|
||
# `!= "day_trade"` 로 탈락하던 문제 + 'daytrading' 누락을 동시에 수정.
|
||
result = await db.execute(
|
||
select(ActiveTrade)
|
||
.where(
|
||
ActiveTrade.stock_code == stock_code,
|
||
ActiveTrade.status == "active",
|
||
not_dt_strategy_clause(ActiveTrade.strategy_name),
|
||
)
|
||
.order_by(ActiveTrade.id.desc())
|
||
.limit(1)
|
||
)
|
||
return result.scalars().first()
|
||
|
||
|
||
async def count_swing_positions(db: AsyncSession) -> int:
|
||
"""현재 스윙 보유 종목 수 반환.
|
||
|
||
swing/entry_service.MAX_SWING_POSITIONS(현재 20) 한도 체크에 사용.
|
||
|
||
Args:
|
||
db: AsyncSession 인스턴스
|
||
|
||
Returns:
|
||
int: 스윙 보유 종목 수
|
||
"""
|
||
from sqlalchemy import func
|
||
result = await db.execute(
|
||
select(func.count(Portfolio.id)).where(
|
||
Portfolio.quantity > 0,
|
||
Portfolio.strategy_type == STRATEGY_TYPE,
|
||
)
|
||
)
|
||
return result.scalar_one() or 0
|
||
|
||
|
||
# ── [2026-10-08] 스윙 전략 진입가 (Portfolio 평단 혼합 방지) ──────────────────
|
||
# Portfolio 행은 stock_code UNIQUE 라 같은 종목의 스윙+DT 수량·평단이 합산된다(§3.5).
|
||
# KIS pchs_avg_pric 는 이동평균법이라 DT 가 전량 매도한 뒤에도 혼합 평단이 그대로 남는다
|
||
# (예: 스윙 5주@68,000 + DT 8주@66,900 → 67,323 — DT 청산 후에도 67,323).
|
||
# 따라서 스윙 손익·손절은 **스윙 소유 ActiveTrade.entry_price** 를 기준으로 한다.
|
||
# ActiveTrade 가 오래된 유령 행이면 진입가가 엉뚱할 수 있으므로 Portfolio 평단과의 괴리가
|
||
# 이 비율을 넘으면 **Portfolio 평단(KIS 실제 매입 평단)** 을 쓰고 warning 을 남긴다.
|
||
#
|
||
# [2026-10-09 하드닝 R1 / 검증 1회차 V1a M-2] 이전에는 괴리 > 30% 면 "높은 값"을 썼다.
|
||
# 유령 entry 100,000 / 실평단 50,000 / 현재가 52,000 이면 -48% 로 판정돼 **허위 전량 손절**
|
||
# + loss_tracker 허위 손실(3연패 → 42일 진입 금지)이 났다. 30% 괴리는 DT 혼합 평단으로는
|
||
# 만들어지기 어렵고(혼합 평단은 두 매입가 사이에 있다) 유령/오염 행일 가능성이 압도적이므로,
|
||
# 실제 돈이 걸린 KIS 평단을 기준으로 한다. 반대 방향(유령 entry 가 낮은 경우)도 KIS 평단이
|
||
# 높으므로 손절이 늦춰지지 않는다 — 계좌 기준 손실이 그대로 손절선에 반영된다.
|
||
SWING_ENTRY_PRICE_MAX_DEVIATION = 0.30
|
||
|
||
|
||
def resolve_swing_entry_price_v2(
|
||
active_trade_entry_price,
|
||
portfolio_avg_price,
|
||
stock_code: str = "",
|
||
) -> tuple[float, str]:
|
||
"""스윙 손익·손절 판정용 진입가 (순수 함수 — DB·네트워크 없음).
|
||
|
||
Returns:
|
||
(price, basis) — basis:
|
||
"active_trade" : 스윙 ActiveTrade.entry_price (정상)
|
||
"portfolio_avg_deviation" : 두 값 괴리 > SWING_ENTRY_PRICE_MAX_DEVIATION →
|
||
Portfolio 평단(KIS 실제 매입 평단). ActiveTrade 는 유령 의심.
|
||
(2026-10-09 이전의 "conservative_max"=max(두 값) 는 폐기 —
|
||
유령 고가 entry 로 허위 전량 손절을 냈다)
|
||
"portfolio_avg" : ActiveTrade 진입가 없음 → Portfolio 평단(혼합 가능) 폴백
|
||
"none" : 둘 다 없음(0) — 호출부가 판정을 건너뛴다
|
||
"""
|
||
try:
|
||
at = float(active_trade_entry_price or 0)
|
||
except (TypeError, ValueError):
|
||
at = 0.0
|
||
try:
|
||
pa = float(portfolio_avg_price or 0)
|
||
except (TypeError, ValueError):
|
||
pa = 0.0
|
||
|
||
if at > 0 and pa > 0:
|
||
dev = abs(at / pa - 1.0)
|
||
if dev <= SWING_ENTRY_PRICE_MAX_DEVIATION + 1e-9: # 부동소수 경계(정확히 30%) 포함
|
||
return at, "active_trade"
|
||
from app.core.log_throttle import warn_throttled
|
||
warn_throttled(
|
||
logger, f"swing_entry_price:{stock_code}",
|
||
f"[SwingEntryPrice] {stock_code}: ActiveTrade 진입가 {at:,.2f} 와 Portfolio 평단 "
|
||
f"{pa:,.2f} 괴리 {dev:.1%} > {SWING_ENTRY_PRICE_MAX_DEVIATION:.0%} — 유령/오염 행 의심, "
|
||
f"KIS 실제 평단 {pa:,.2f} 사용(ActiveTrade 수동 확인 필요)"
|
||
)
|
||
return pa, "portfolio_avg_deviation"
|
||
if at > 0:
|
||
return at, "active_trade"
|
||
if pa > 0:
|
||
return pa, "portfolio_avg"
|
||
return 0.0, "none"
|
||
|
||
|
||
# ── [2026-10-09 하드닝 R1] 스윙 SELL 의 출처(시스템 진입 vs 동기화 유입) 분리 ─────────
|
||
# 운영 실측: swing SELL 중 선행 스윙 BUY 가 없는 매도 35건(pnl +62,789) vs 스윙 BUY 이후
|
||
# 매도 30건(pnl -166,885). 유입분이 sync 로 'swing' 태그를 받아 청산되면 전략 성과에 섞여
|
||
# 시스템 진입 신호의 품질이 가려진다. 통계 API 가 쓸 수 있도록 **원장 순수량** 기준으로
|
||
# 매도 수량을 system/inflow 로 배분하는 순수 함수를 둔다(DB·네트워크 없음).
|
||
_SELL_ORIGIN_EXCLUDED_STATUSES = frozenset({"FAILED", "CANCELLED", "CANCELED", "REJECTED"})
|
||
|
||
|
||
def _row_get_v2(row, key, default=None):
|
||
if isinstance(row, dict):
|
||
return row.get(key, default)
|
||
return getattr(row, key, default)
|
||
|
||
|
||
def split_swing_sells_by_origin_v2(trades) -> list[dict]:
|
||
"""스윙 TradeHistory 행들을 시간순으로 훑어 각 SELL 의 수량을 출처별로 나눈다.
|
||
|
||
규칙: 종목별 '시스템 보유' = 누적 스윙 BUY 체결수량 − 이미 system 으로 배분된 SELL 수량.
|
||
SELL 은 시스템 보유분부터 소진하고(FIFO 무관, 수량만), 초과분은 inflow(동기화 유입)로 본다.
|
||
strategy_type 이 swing(또는 NULL) 이 아닌 행, 실패·취소·거부 행은 무시한다.
|
||
|
||
Returns:
|
||
SELL 마다 {id, stock_code, created_at, sell_qty, system_qty, inflow_qty,
|
||
origin: "system"|"inflow"|"mixed"} (입력 시간순)
|
||
"""
|
||
rows = []
|
||
for idx, t in enumerate(trades or []):
|
||
st = _row_get_v2(t, "strategy_type")
|
||
if st not in (None, STRATEGY_TYPE):
|
||
continue
|
||
status = str(_row_get_v2(t, "status") or "").upper()
|
||
if status in _SELL_ORIGIN_EXCLUDED_STATUSES:
|
||
continue
|
||
side = str(_row_get_v2(t, "order_type") or "").upper()
|
||
if side not in ("BUY", "SELL"):
|
||
continue
|
||
q = _row_get_v2(t, "executed_quantity")
|
||
if q is None:
|
||
q = _row_get_v2(t, "order_quantity")
|
||
try:
|
||
q = int(q or 0)
|
||
except (TypeError, ValueError):
|
||
q = 0
|
||
if q <= 0:
|
||
continue
|
||
rows.append((_row_get_v2(t, "created_at"), idx, side, q, t))
|
||
|
||
# created_at 없는 행은 입력 순서를 유지(idx 보조키)
|
||
rows.sort(key=lambda r: (r[0] is None, r[0] if r[0] is not None else 0, r[1]))
|
||
|
||
held: dict[str, int] = {}
|
||
out: list[dict] = []
|
||
for created_at, _idx, side, q, t in rows:
|
||
code = _row_get_v2(t, "stock_code")
|
||
if side == "BUY":
|
||
held[code] = held.get(code, 0) + q
|
||
continue
|
||
sys_q = min(q, held.get(code, 0))
|
||
held[code] = held.get(code, 0) - sys_q
|
||
inflow_q = q - sys_q
|
||
origin = "system" if inflow_q == 0 else ("inflow" if sys_q == 0 else "mixed")
|
||
out.append({
|
||
"id": _row_get_v2(t, "id"),
|
||
"stock_code": code,
|
||
"created_at": created_at,
|
||
"sell_qty": q,
|
||
"system_qty": sys_q,
|
||
"inflow_qty": inflow_q,
|
||
"origin": origin,
|
||
})
|
||
return out
|