auto-stock/backend/app/services/system_buy_registry.py

"""system_buy_registry — "시스템이 방금 이 종목을 매수했다"는 **즉시(인메모리) 근거**.

## 왜 필요한가 (유입 잔고 강등 레이스)

`portfolio.sync_portfolio()` 는 KIS 잔고 1행의 `strategy_type` 을 판정할 때
"시스템 매수 근거"를 **DB**(`TradeHistory` BUY ∪ 시딩 아닌 `ActiveTrade`)에서 찾는다.
그런데 두 테이블은 **둘 다 db_worker 큐를 거쳐 비동기로 INSERT** 된다(CLAUDE.md §3.2).
게다가 매수 직후 `trade_executor._verify_and_sync_execution()` 이
`_delayed_sync_portfolio(0)` 로 **즉시** sync 를 돌린다.

    매수 체결 → Portfolio 행(swing) → [db_worker INSERT 전] → sync 실행
      → DB 근거 없음 → external 강등 → 손절·익절·트레일링이 전부 안 걸림

즉 **방금 산 정상 포지션이 리스크 관리 밖으로 떨어진다.** 이 모듈은 그 창을 막는다 —
주문 성공 시점에 호출부가 `note_system_buy()` 로 종목을 등록하면, DB 반영 전이라도
sync 가 `snapshot_codes()` 로 근거를 본다.

## 불변식

**정상 포지션은 강등되지 않는다.** 이 모듈은 근거를 *추가*만 하며 제거하지 않는다.
따라서 오탐(잘못된 강등)은 못 만들고, 최악의 경우 유입 잔고 격리가 TTL 만큼
늦어질 뿐이다(격리 지연은 허용 가능, 오강등은 허용 불가).

## ⚠️ 용도 제한 — 승격 근거로 쓰지 말 것 (M-2, 2026-09-30)

이 레지스트리가 쓰여도 되는 곳은 **두 군데뿐**이다:
  ① `portfolio._should_demote_to_external()` 의 1차 방어 (강등을 **막는** 근거)
  ② 신규 Portfolio 행의 초기 `strategy_type` 태깅 (원장 INSERT 전이라 필요)

**기존 `external` 행의 승격 판정에는 쓸 수 없다.** 두 용도의 안전 방향이 반대다 —
강등 방어는 "근거가 있다고 보는 것"이 안전하고, 승격은 "없다고 보는 것"이 안전하다.
실제 사고 경로: 유입 3주 + DT 3주 행에서 DT 가 15:20 EOD 전량 청산하면 원장 net·
비DT active·DT 인메모리가 모두 0 이 되는데 **이 레지스트리만 TTL 내라** `_sys=3=
행수량` 이 되어 승격되고 `strategy_type='swing'` 이 **DB 에 영구 기록**된다. 장이
닫혀 다음 sync 가 오지 않으면 그대로 밤을 넘기고 다음 날 개장 시 유입분이 자동 손절
대상이 된다. 그래서 `sync_portfolio` 는 근거를 `_system_*_durable`(레지스트리 제외)과
`_system_*`(포함) 두 벌로 만들고 **승격은 durable 만** 본다.

`note_system_buy()` 는 **매도를 반영하지 않는다** — 그래서 이 모듈의 수량은 "방금
주문한 양"이지 "현재 보유량"이 아니다. 위 용도 제한을 지키면 그 차이가 무해하다.

## 주의

- 프로세스 재기동으로 비워진다. 그 경우는 이미 DB 에 근거가 들어가 있고,
  `portfolio.py` 쪽 "연속 확인 + 최소 경과시간" 게이트가 2차 방어로 남는다.
- 실시간 계층에서 호출되므로 **DB I/O 없음**(CLAUDE.md §3.1).
- TTL 은 db_worker 최대 지연(지수 백오프 1+2+4초 ×3회 + DLQ 재시도)을 넉넉히 덮는 값.
"""
import logging
import time

logger = logging.getLogger(__name__)

# 등록 유효 시간(초). db_worker 큐 지연·재시도(최대 수십 초)를 넉넉히 덮는다.
# 길게 잡아도 부작용은 "유입 잔고 격리가 최대 이 시간만큼 늦어짐" 뿐이다.
SYSTEM_BUY_TTL_SEC = 900

# {stock_code: [monotonic 등록 시각, 누적 주문 수량]}
_recent_buys: dict = {}


def note_system_buy(stock_code: str, quantity: int = 0) -> None:
    """시스템 매수 주문이 나갔음을 즉시 기록 (동기 함수 — 어디서나 호출 가능).

    Args:
        stock_code: 종목코드
        quantity: 주문 수량. 부분 혼재 판정(`_classify_holding_strategy_type_v2`)이
            **수량**을 쓰므로 함께 남긴다. 0 이면 "수량 미상" 이고, 그 경우 호출부는
            근거 수량을 원장에서만 얻는다. TTL 안에서 여러 번 호출되면 **합산**한다
            (분할 주문·추가매수).

    실패해도 매매를 막아서는 안 되므로 절대 예외를 올리지 않는다.
    """
    if not stock_code:
        return
    try:
        _c = str(stock_code)
        _q = max(0, int(quantity or 0))
        st = _recent_buys.get(_c)
        if st is None or time.monotonic() - st[0] > SYSTEM_BUY_TTL_SEC:
            _recent_buys[_c] = [time.monotonic(), _q]
        else:
            st[0] = time.monotonic()
            st[1] += _q
    except Exception as e:  # pragma: no cover - 방어적
        logger.warning(f"[system_buy_registry] 등록 실패 ({stock_code}): {e}")


def snapshot_quantities() -> dict:
    """TTL 내 시스템 매수 **수량** {code: qty} (만료분 정리 포함).

    `sync_portfolio` 가 `_system_qty` 의 하한으로 합성한다 — 원장 INSERT 전에도
    "이 행의 N주는 시스템 몫" 이라는 사실이 보이게 해, 부분 혼재 판정이 방금 산
    수량을 유입분으로 오인하지 않게 한다.
    """
    now = time.monotonic()
    for c in [c for c, st in _recent_buys.items() if now - st[0] > SYSTEM_BUY_TTL_SEC]:
        _recent_buys.pop(c, None)
    return {c: int(st[1]) for c, st in _recent_buys.items()}


def has_recent_system_buy(stock_code: str) -> bool:
    """해당 종목에 TTL 내 시스템 매수 기록이 있는가."""
    if not stock_code:
        return False
    st = _recent_buys.get(str(stock_code))
    if st is None:
        return False
    if time.monotonic() - st[0] > SYSTEM_BUY_TTL_SEC:
        _recent_buys.pop(str(stock_code), None)
        return False
    return True


def snapshot_codes() -> set:
    """TTL 내 시스템 매수 종목 집합 (만료분 정리 포함).

    `portfolio._classify_holding_strategy_type(system_codes=...)` 에 합집합으로 넣는다.
    """
    now = time.monotonic()
    expired = [c for c, st in _recent_buys.items() if now - st[0] > SYSTEM_BUY_TTL_SEC]
    for c in expired:
        _recent_buys.pop(c, None)
    return set(_recent_buys)


def clear() -> None:
    """테스트용 초기화."""
    _recent_buys.clear()