51 lines
1.3 KiB
Python
51 lines
1.3 KiB
Python
"""조건검색 공통 — 브로커 무관 필수만.
|
|
|
|
키움 REAL / LS AFR 시드·프로토콜은 여기 넣지 않는다 (분기 유발).
|
|
멤버십 반영은 ``ConditionSearchManager._apply_result`` 가 본체.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Dict, Iterable, List, Optional
|
|
|
|
# LS: N/R/O · 키움: I/D (키움은 재진입 플래그 없음 → I 재수신)
|
|
JOB_KO = {
|
|
"N": "진입",
|
|
"R": "재진입",
|
|
"O": "이탈",
|
|
"I": "진입",
|
|
"D": "이탈",
|
|
}
|
|
|
|
|
|
def normalize_job_flag(flag: Optional[str]) -> str:
|
|
return str(flag or "").strip().upper()
|
|
|
|
|
|
def job_kind(flag: Optional[str]) -> Optional[str]:
|
|
"""enter | reenter | exit | None."""
|
|
j = normalize_job_flag(flag)
|
|
if j in ("N", "I"):
|
|
return "enter"
|
|
if j == "R":
|
|
return "reenter"
|
|
if j in ("O", "D"):
|
|
return "exit"
|
|
return None
|
|
|
|
|
|
def rows_for_codes(
|
|
codes: Iterable[str],
|
|
names: Optional[Dict[str, str]] = None,
|
|
) -> List[Dict[str, str]]:
|
|
"""``_apply_result`` 입력 형태 (code/name)."""
|
|
nm = names or {}
|
|
out: List[Dict[str, str]] = []
|
|
seen = set()
|
|
for c in codes:
|
|
code = str(c or "").strip()
|
|
if not code or code in seen:
|
|
continue
|
|
seen.add(code)
|
|
out.append({"code": code, "name": str(nm.get(code) or code)})
|
|
return out
|