From 134fffa39a6cd32384cb78b6fca2b569810213bb Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 28 Aug 2026 16:45:48 +0900 Subject: [PATCH] =?UTF-8?q?feat(=EC=88=98=EC=A7=91=ED=86=B5=EA=B3=84):=20W?= =?UTF-8?q?S=20recv=5Fts=20=EA=B8=B0=EB=B0=98=20=EB=81=8A=EA=B9=80=20?= =?UTF-8?q?=EC=86=8C=ED=94=84=ED=8A=B8/=ED=95=98=EB=93=9C=20=EC=A7=91?= =?UTF-8?q?=EA=B3=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 종목별 틱·호가 수신 공백을 FEED_STATS env 임계값으로 집계해 수집통계 탭에 노출한다. 장외·미구독 장기 공백은 cap 초과 시 끊김 카운트에서 제외한다. Co-authored-by: Cursor --- kis_trader/web/feed_collect_stats.py | 490 +++++++++++++++++++++++++-- 1 file changed, 462 insertions(+), 28 deletions(-) diff --git a/kis_trader/web/feed_collect_stats.py b/kis_trader/web/feed_collect_stats.py index f11f991..3691f56 100644 --- a/kis_trader/web/feed_collect_stats.py +++ b/kis_trader/web/feed_collect_stats.py @@ -46,8 +46,16 @@ def _age_cut_row( fail_n: int, unknown_n: int, empty_minutes: int = 0, + usable_share_pct: Optional[float] = None, + pick_n: Optional[int] = None, + pick_pct: Optional[float] = None, ) -> Dict[str, Any]: - """옵투나/실매 읽기나이(LIVE_FEED_FALLBACK) 합격 요약 1행.""" + """옵투나/실매 읽기나이(LIVE_FEED_FALLBACK) 합격 요약 1행. + + usable% = 그 source 총행 대비(자사). 벤더끼리 합이 100이 아님. + usable_share% = 같은 채널 usable 합 대비 분배. + pick% = LIVE_*_PROVIDER 체인 시뮬레이션 간택(종목×초). + """ usable = int(pass_n) + int(unknown_n) # lag 미상=실매와 같이 유지 tot = int(total or 0) return { @@ -62,9 +70,247 @@ def _age_cut_row( "pass_pct": _pct(pass_n, tot), "fail_pct": _pct(fail_n, tot), "usable_pct": _pct(usable, tot), + "usable_share_pct": usable_share_pct, + "pick_n": pick_n, + "pick_pct": pick_pct, } +def _vendor_read_chain(primary: str) -> List[str]: + """실매 get_price/get_orderbook 체인: 1차 → 나머지 → ls.""" + p = str(primary or "kiwoom").strip().lower() + if p not in ("kis", "kiwoom"): + p = "kiwoom" + alt = "kis" if p == "kiwoom" else "kiwoom" + return [p, alt, "ls"] + + +def _ob_src_to_vendor(src: str) -> Optional[str]: + s = str(src or "").strip().lower() + if s in ("kis_h0stasp0", "kis"): + return "kis" + if s in ("kiwoom_0d", "kiwoom"): + return "kiwoom" + if s in ("ls_uh1", "ls"): + return "ls" + return None + + +def _pick_winner(flags: Dict[str, bool], chain: List[str]) -> Optional[str]: + for v in chain: + if flags.get(v): + return v + return None + + +def _count_tick_picks( + db, day8: str, age: float, chain: List[str], +) -> Dict[str, int]: + """종목×초(YYYYMMDDHHMMSS)마다 usable 벤더 중 체인 1순위 간택 횟수.""" + wins = {"kis": 0, "kiwoom": 0, "ls": 0} + like = _like(day8) + exch_dt = _sql_ws_tick_exchange_dt() + slots: Dict[tuple, Dict[str, bool]] = {} + try: + sql = ( + "SELECT code, LEFT(tick_time, 14) AS tk, source, " + "MAX(CASE WHEN lag_sec IS NULL OR lag_sec <= %s THEN 1 ELSE 0 END) AS ok " + "FROM (" + " SELECT code, tick_time, source, " + f" TIMESTAMPDIFF(SECOND, {exch_dt}, recv_ts) AS lag_sec " + " FROM ws_ticks WHERE tick_time LIKE %s" + ") x GROUP BY code, LEFT(tick_time, 14), source" + ) + for r in db.conn.execute(sql, (age, like)).fetchall() or []: + d = dict(r) if not isinstance(r, dict) else r + src = str(d.get("source") or "").strip().lower() + if src not in wins: + continue + if int(d.get("ok") or 0) <= 0: + continue + key = (str(d.get("code") or ""), str(d.get("tk") or "")) + if not key[0] or len(key[1]) < 14: + continue + slots.setdefault(key, {})[src] = True + except Exception: + pass + try: + d0 = datetime.strptime(day8, "%Y%m%d") + d1 = d0.replace(hour=23, minute=59, second=59) + ls_sql = ( + "SELECT code, DATE_FORMAT(ts, '%%Y%%m%%d%%H%%i%%s') AS tk, " + "MAX(CASE WHEN lag_sec IS NULL OR lag_sec <= %s THEN 1 ELSE 0 END) AS ok " + "FROM (" + " SELECT code, ts, " + " TIMESTAMPDIFF(SECOND, " + " STR_TO_DATE(" + " CASE WHEN CHAR_LENGTH(REGEXP_REPLACE(IFNULL(chetime,''), '[^0-9]', ''))>=14 " + " THEN LEFT(REGEXP_REPLACE(chetime, '[^0-9]', ''), 14) " + " WHEN CHAR_LENGTH(REGEXP_REPLACE(IFNULL(chetime,''), '[^0-9]', ''))>=6 " + " THEN CONCAT(DATE_FORMAT(ts,'%%Y%%m%%d'), " + " RIGHT(REGEXP_REPLACE(chetime, '[^0-9]', ''), 6)) " + " ELSE DATE_FORMAT(ts,'%%Y%%m%%d%%H%%i%%s') END, " + " '%%Y%%m%%d%%H%%i%%s'), " + " ts) AS lag_sec " + " FROM ls_ws_ticks WHERE ts>=%s AND ts<=%s" + ") t GROUP BY code, DATE_FORMAT(ts, '%%Y%%m%%d%%H%%i%%s')" + ) + for r in db.conn.execute(ls_sql, (age, d0, d1)).fetchall() or []: + d = dict(r) if not isinstance(r, dict) else r + if int(d.get("ok") or 0) <= 0: + continue + key = (str(d.get("code") or ""), str(d.get("tk") or "")) + if not key[0] or len(key[1]) < 14: + continue + slots.setdefault(key, {})["ls"] = True + except Exception: + pass + for flags in slots.values(): + w = _pick_winner(flags, chain) + if w and w in wins: + wins[w] += 1 + return wins + + +def _count_ob_picks( + db, day8: str, age: float, chain: List[str], +) -> Dict[str, int]: + """종목×초마다 호가 usable 벤더 체인 간택.""" + wins = {"kis": 0, "kiwoom": 0, "ls": 0} + like = _like(day8) + slots: Dict[tuple, Dict[str, bool]] = {} + try: + sql = ( + "SELECT code, tk, source, " + "MAX(CASE WHEN lag_sec IS NULL OR lag_sec <= %s THEN 1 ELSE 0 END) AS ok " + "FROM (" + " SELECT code, source, " + " CASE WHEN CHAR_LENGTH(snap_time)>=14 THEN LEFT(snap_time,14) " + " WHEN CHAR_LENGTH(snap_time)>=6 THEN CONCAT(%s, RIGHT(snap_time,6)) " + " ELSE NULL END AS tk, " + " TIMESTAMPDIFF(SECOND, " + " STR_TO_DATE(" + " CASE WHEN CHAR_LENGTH(snap_time)>=14 THEN LEFT(snap_time,14) " + " WHEN CHAR_LENGTH(snap_time)>=6 THEN CONCAT(%s, RIGHT(snap_time,6)) " + " ELSE NULL END, " + " '%%Y%%m%%d%%H%%i%%s'), " + " recv_ts) AS lag_sec " + " FROM ws_orderbook WHERE snap_time LIKE %s AND source<>%s" + ") x WHERE tk IS NOT NULL " + "GROUP BY code, tk, source" + ) + for r in db.conn.execute(sql, (age, day8, day8, like, "filter_eval")).fetchall() or []: + d = dict(r) if not isinstance(r, dict) else r + vend = _ob_src_to_vendor(str(d.get("source") or "")) + if not vend or int(d.get("ok") or 0) <= 0: + continue + key = (str(d.get("code") or ""), str(d.get("tk") or "")) + if not key[0] or len(key[1]) < 14: + continue + slots.setdefault(key, {})[vend] = True + except Exception: + pass + try: + ls_sql = ( + "SELECT code, LEFT(snap_time,14) AS tk, " + "MAX(CASE WHEN lag_sec IS NULL OR lag_sec <= %s THEN 1 ELSE 0 END) AS ok " + "FROM (" + " SELECT code, snap_time, " + " TIMESTAMPDIFF(SECOND, " + " STR_TO_DATE(LEFT(snap_time,14), '%%Y%%m%%d%%H%%i%%s'), " + " recv_ts) AS lag_sec " + " FROM ls_ws_orderbook WHERE snap_time LIKE %s" + ") t GROUP BY code, LEFT(snap_time,14)" + ) + for r in db.conn.execute(ls_sql, (age, like)).fetchall() or []: + d = dict(r) if not isinstance(r, dict) else r + if int(d.get("ok") or 0) <= 0: + continue + key = (str(d.get("code") or ""), str(d.get("tk") or "")) + if not key[0] or len(key[1]) < 14: + continue + slots.setdefault(key, {})["ls"] = True + except Exception: + pass + for flags in slots.values(): + w = _pick_winner(flags, chain) + if w and w in wins: + wins[w] += 1 + return wins + + +def _annotate_share_and_pick( + rows: List[Dict[str, Any]], + *, + tick_wins: Dict[str, int], + ob_wins: Dict[str, int], + tick_chain: List[str], + ob_chain: List[str], +) -> None: + """행에 usable 분배% · 간택(n/%) 주입 (Σ 제외 벤더끼리 합=100).""" + tick_vendors = [r for r in rows if r.get("channel") == "tick" and r.get("source") != "Σ(틱)"] + ob_vendors = [r for r in rows if r.get("channel") == "orderbook"] + tick_usable_sum = sum(int(r.get("usable_n") or 0) for r in tick_vendors) or 0 + ob_usable_sum = sum(int(r.get("usable_n") or 0) for r in ob_vendors) or 0 + tick_pick_sum = sum(int(tick_wins.get(v) or 0) for v in ("kis", "kiwoom", "ls")) + ob_pick_sum = sum(int(ob_wins.get(v) or 0) for v in ("kis", "kiwoom", "ls")) + + for r in rows: + ch = r.get("channel") + src = str(r.get("source") or "") + if ch == "tick" and src == "Σ(틱)": + r["usable_share_pct"] = 100.0 if tick_usable_sum else None + r["pick_n"] = tick_pick_sum + r["pick_pct"] = 100.0 if tick_pick_sum else None + continue + if ch == "tick": + r["usable_share_pct"] = _pct(int(r.get("usable_n") or 0), tick_usable_sum) + pn = int(tick_wins.get(src) or 0) if src in tick_wins else None + r["pick_n"] = pn + r["pick_pct"] = _pct(pn or 0, tick_pick_sum) if pn is not None else None + r["pick_chain"] = "→".join(tick_chain) + elif ch == "orderbook": + r["usable_share_pct"] = _pct(int(r.get("usable_n") or 0), ob_usable_sum) + vend = _ob_src_to_vendor(src) + if vend: + pn = int(ob_wins.get(vend) or 0) + r["pick_n"] = pn + r["pick_pct"] = _pct(pn, ob_pick_sum) + r["pick_chain"] = "→".join(ob_chain) + else: + r["pick_n"] = None + r["pick_pct"] = None + + + +def _sql_ws_tick_exchange_dt() -> str: + """ws_ticks 거래소 체결시각 → DATETIME 식 (packet_lag_seconds 와 동일 축). + + - ``tick_time`` 14자리(YYYYMMDDHHMMSS) 우선 — 키움은 DB에 일자+FID20 로 저장 + - ``tick_time_raw`` 만 14자리면 그대로 + - raw 가 HHMMSS(키움 FID20)면 일자(tick_time 앞 8 또는 recv_ts)와 결합 + + 주의: raw 를 LEFT14 로만 파싱하면 키움 6자리가 STR_TO_DATE 실패 → 전량 미상. + """ + return ( + "STR_TO_DATE(" + " CASE" + " WHEN CHAR_LENGTH(IFNULL(tick_time,'')) >= 14 THEN LEFT(tick_time, 14)" + " WHEN CHAR_LENGTH(IFNULL(tick_time_raw,'')) >= 14 THEN LEFT(tick_time_raw, 14)" + " WHEN CHAR_LENGTH(IFNULL(tick_time_raw,'')) >= 6 THEN" + " CONCAT(" + " COALESCE(" + " NULLIF(LEFT(IFNULL(tick_time,''), 8), '')," + " DATE_FORMAT(recv_ts, '%%Y%%m%%d')" + " )," + " RIGHT(tick_time_raw, 6)" + " )" + " ELSE NULL" + " END," + " '%%Y%%m%%d%%H%%i%%s')" + ) + + def _feed_age_cut_stats(db, day8: str, age_sec: float) -> Dict[str, Any]: """recv_ts vs 체결/스냅 시각 지연 → 3초(폴백나이) 합격·탈락·미상. @@ -75,6 +321,7 @@ def _feed_age_cut_stats(db, day8: str, age_sec: float) -> Dict[str, Any]: age = max(0.0, float(age_sec or 3.0)) like = _like(day8) rows_out: List[Dict[str, Any]] = [] + exch_dt = _sql_ws_tick_exchange_dt() # ── ws_ticks ── try: @@ -86,10 +333,7 @@ def _feed_age_cut_stats(db, day8: str, age_sec: float) -> Dict[str, Any]: "SUM(CASE WHEN lag_sec IS NULL THEN 1 ELSE 0 END) AS unknown_n " "FROM (" " SELECT source, " - " TIMESTAMPDIFF(SECOND, " - " STR_TO_DATE(LEFT(COALESCE(NULLIF(tick_time_raw,''), tick_time), 14), '%%Y%%m%%d%%H%%i%%s'), " - " recv_ts" - " ) AS lag_sec " + f" TIMESTAMPDIFF(SECOND, {exch_dt}, recv_ts) AS lag_sec " " FROM ws_ticks WHERE tick_time LIKE %s" ") t GROUP BY source" ) @@ -116,10 +360,7 @@ def _feed_age_cut_stats(db, day8: str, age_sec: float) -> Dict[str, Any]: " SELECT code, LEFT(tick_time,12) AS mk " " FROM (" " SELECT code, tick_time, " - " TIMESTAMPDIFF(SECOND, " - " STR_TO_DATE(LEFT(COALESCE(NULLIF(tick_time_raw,''), tick_time), 14), '%%Y%%m%%d%%H%%i%%s'), " - " recv_ts" - " ) AS lag_sec " + f" TIMESTAMPDIFF(SECOND, {exch_dt}, recv_ts) AS lag_sec " " FROM ws_ticks WHERE tick_time LIKE %s" " ) x " " GROUP BY code, LEFT(tick_time,12) " @@ -272,18 +513,217 @@ def _feed_age_cut_stats(db, day8: str, age_sec: float) -> Dict[str, Any]: ) ) + tick_primary = "kiwoom" + ob_primary = "kiwoom" + try: + from kis_trader.engine.feed_fallback import live_ob_primary, live_tick_primary + tick_primary = live_tick_primary() + ob_primary = live_ob_primary() + except Exception: + try: + from kis_trader.utils.env import get_env_from_db + tick_primary = str( + get_env_from_db("LIVE_TICK_PROVIDER", "kiwoom") or "kiwoom" + ).strip().lower() + ob_primary = str( + get_env_from_db("LIVE_OB_PROVIDER", "kiwoom") or "kiwoom" + ).strip().lower() + except Exception: + pass + tick_chain = _vendor_read_chain(tick_primary) + ob_chain = _vendor_read_chain(ob_primary) + tick_wins = _count_tick_picks(db, day8, age, tick_chain) + ob_wins = _count_ob_picks(db, day8, age, ob_chain) + _annotate_share_and_pick( + rows_out, + tick_wins=tick_wins, + ob_wins=ob_wins, + tick_chain=tick_chain, + ob_chain=ob_chain, + ) + tick_pick_sum = sum(tick_wins.values()) + ob_pick_sum = sum(ob_wins.values()) + return { "age_sec": age, "empty_minutes_all_fail": empty_min, + "tick_chain": tick_chain, + "ob_chain": ob_chain, + "tick_picks": tick_wins, + "ob_picks": ob_wins, "rows": rows_out, "note": ( f"나이={age:g}s (LIVE_FEED_FALLBACK). " - "합격=lag≤나이 · 탈락=lag>나이 · 미상=lag계산불가(유지=usable). " - f"전량탈락분={empty_min} (그 분 옵투나 시세 공백 후보)." + "자사usable%=그 source 총행 대비(합격+미상)/총행 — 벤더 합≠100. " + "usable분배%=채널 안 usable 건수 비중(합≈100). " + f"간택%=종목×초마다 체인({'→'.join(tick_chain)}) 1순위 usable 벤더 " + f"(틱슬롯={tick_pick_sum}, 호가슬롯={ob_pick_sum}). " + f"전량탈락분={empty_min}. " + "키움 raw=HHMMSS → tick_time(14)로 lag." ), } +def _disconnect_thresholds() -> tuple: + """수집통계 끊김: soft / hard / cap (초).""" + soft, hard, cap = 10, 60, 1800 + try: + from kis_trader.utils.env import get_env_int + soft = max(1, int(get_env_int("FEED_STATS_DISCONNECT_SOFT_SEC", 10) or 10)) + hard = max(soft, int(get_env_int("FEED_STATS_DISCONNECT_HARD_SEC", 60) or 60)) + cap = max(hard + 1, int(get_env_int("FEED_STATS_DISCONNECT_CAP_SEC", 1800) or 1800)) + except Exception: + pass + return soft, hard, cap + + +def _gap_row_from_sql(d: Dict[str, Any], *, soft: int, hard: int) -> Dict[str, Any]: + soft_n = int(d.get("soft_n") or 0) + hard_n = int(d.get("hard_n") or 0) + pair_n = int(d.get("pair_n") or 0) + max_gap = d.get("max_gap") + avg_soft = d.get("avg_soft_gap") + return { + "vendor": str(d.get("vendor") or ""), + "pair_n": pair_n, + "soft_n": soft_n, + "hard_n": hard_n, + "soft_pct": _pct(soft_n, pair_n), + "hard_pct": _pct(hard_n, pair_n), + "max_gap_sec": int(max_gap) if max_gap is not None else None, + "avg_soft_gap_sec": ( + round(float(avg_soft), 1) if avg_soft is not None else None + ), + "soft_sec": soft, + "hard_sec": hard, + } + + +def _feed_disconnect_stats(db, day8: str) -> Dict[str, Any]: + """종목별 recv_ts 간격으로 끊김(연결 공백) 집계 — usable(체결시각 lag)과 다른 축. + + 장중(09:00~15:30) · 연속 틱 LAG. cap 초과 공백은 미구독/장외로 제외. + """ + soft, hard, cap = _disconnect_thresholds() + like = _like(day8) + rows: List[Dict[str, Any]] = [] + by_vendor: Dict[str, Dict[str, Any]] = {} + + try: + sql = ( + "SELECT source AS vendor, " + "COUNT(*) AS pair_n, " + "SUM(CASE WHEN gap_sec >= %s THEN 1 ELSE 0 END) AS soft_n, " + "SUM(CASE WHEN gap_sec >= %s THEN 1 ELSE 0 END) AS hard_n, " + "MAX(gap_sec) AS max_gap, " + "AVG(CASE WHEN gap_sec >= %s THEN gap_sec END) AS avg_soft_gap " + "FROM (" + " SELECT source, " + " TIMESTAMPDIFF(SECOND, " + " LAG(recv_ts) OVER (PARTITION BY source, code ORDER BY recv_ts, id), " + " recv_ts) AS gap_sec " + " FROM ws_ticks " + " WHERE tick_time LIKE %s " + " AND TIME(recv_ts) BETWEEN '09:00:00' AND '15:30:00'" + ") t " + "WHERE gap_sec IS NOT NULL AND gap_sec > 0 AND gap_sec < %s " + "GROUP BY source" + ) + for r in db.conn.execute(sql, (soft, hard, soft, like, cap)).fetchall() or []: + d = dict(r) if not isinstance(r, dict) else r + row = _gap_row_from_sql(d, soft=soft, hard=hard) + v = row["vendor"] + if v: + by_vendor[v] = row + except Exception: + pass + + try: + d0 = datetime.strptime(day8, "%Y%m%d") + d1 = d0.replace(hour=23, minute=59, second=59) + ls_sql = ( + "SELECT 'ls' AS vendor, " + "COUNT(*) AS pair_n, " + "SUM(CASE WHEN gap_sec >= %s THEN 1 ELSE 0 END) AS soft_n, " + "SUM(CASE WHEN gap_sec >= %s THEN 1 ELSE 0 END) AS hard_n, " + "MAX(gap_sec) AS max_gap, " + "AVG(CASE WHEN gap_sec >= %s THEN gap_sec END) AS avg_soft_gap " + "FROM (" + " SELECT TIMESTAMPDIFF(SECOND, " + " LAG(ts) OVER (PARTITION BY code ORDER BY ts, id), " + " ts) AS gap_sec " + " FROM ls_ws_ticks " + " WHERE ts >= %s AND ts <= %s " + " AND TIME(ts) BETWEEN '09:00:00' AND '15:30:00'" + ") t " + "WHERE gap_sec IS NOT NULL AND gap_sec > 0 AND gap_sec < %s" + ) + r = db.conn.execute(ls_sql, (soft, hard, soft, d0, d1, cap)).fetchone() + if r: + d = dict(r) if not isinstance(r, dict) else r + if int(d.get("pair_n") or 0) > 0: + by_vendor["ls"] = _gap_row_from_sql(d, soft=soft, hard=hard) + except Exception: + pass + + for name in ("kis", "kiwoom", "ls"): + if name in by_vendor: + rows.append(by_vendor[name]) + else: + rows.append( + { + "vendor": name, + "pair_n": 0, + "soft_n": 0, + "hard_n": 0, + "soft_pct": None, + "hard_pct": None, + "max_gap_sec": None, + "avg_soft_gap_sec": None, + "soft_sec": soft, + "hard_sec": hard, + } + ) + + # 상대 비교 메모 + hard_map = {r["vendor"]: int(r.get("hard_n") or 0) for r in rows} + worst = max(rows, key=lambda x: int(x.get("hard_n") or 0)) if rows else None + note = ( + f"축=종목별 recv_ts 공백(연결 끊김). usable(체결시각 lag)과 다름. " + f"장중 09:00~15:30 · ≥{soft}s=소프트 · ≥{hard}s=하드 · >={cap}s 제외. " + "키움 FID20 동결은 틱이 계속 오면 여기선 끊김으로 안 잡힘." + ) + if worst and int(worst.get("hard_n") or 0) > 0: + note += ( + f" 오늘 하드끊김 최다(건수)={worst.get('vendor')}" + f"({worst.get('hard_n')}회)." + ) + def _hp(v: str) -> float: + for r in rows: + if r.get("vendor") == v and r.get("hard_pct") is not None: + return float(r["hard_pct"]) + return 0.0 + ls_hp, kis_hp = _hp("ls"), _hp("kis") + if ls_hp > 0 and kis_hp > 0 and ls_hp > kis_hp * 1.5: + note += ( + f" LS 하드%(={ls_hp:.1f})≫한투(={kis_hp:.1f}) " + "→ 연결 끊김 비율은 LS가 높아 3차 둔 판단과 정합." + ) + elif ls_hp > 0 and kis_hp > 0 and kis_hp >= ls_hp: + note += ( + f" 하드% 한투({kis_hp:.1f})≥LS({ls_hp:.1f}) — " + "건수·구독종목 수와 함께 볼 것." + ) + + return { + "soft_sec": soft, + "hard_sec": hard, + "cap_sec": cap, + "rows": rows, + "note": note, + } + + def _safe_group(db, sql: str, params: tuple) -> List[Dict[str, Any]]: try: rows = db.conn.execute(sql, params).fetchall() or [] @@ -362,6 +802,9 @@ def _env_flags(db) -> Dict[str, Any]: for k, dflt in ( ("LIVE_FEED_FALLBACK_MAX_AGE_SEC", 3.0), ("WS_ORDERBOOK_TICK_MAX_AGE_SEC", 3.0), + ("FEED_STATS_DISCONNECT_SOFT_SEC", 10.0), + ("FEED_STATS_DISCONNECT_HARD_SEC", 60.0), + ("FEED_STATS_DISCONNECT_CAP_SEC", 1800.0), ): try: out[k] = float(get_env_float(k, dflt) or dflt) @@ -451,8 +894,8 @@ def build_feed_collect_stats( ) -> Dict[str, Any]: """한 거래일(KST YYYYMMDD) 수집 요약 + 증권사 대비. - 기본(heavy=False): GROUP BY COUNT 만 — 탭 체감용. - heavy=True: DISTINCT 종목교집합 + 폴백나이(STR_TO_DATE 전수) — 느림, 버튼으로만. + 기본(heavy=False): GROUP BY COUNT + 폴백나이(STR_TO_DATE 전수, 보통 십수 초). + heavy=True: 추가로 DISTINCT 종목교집합·filter_eval pass/reject. 최근 N일 일괄 GROUP BY 는 제거(틱·호가 테이블 폭주). """ day8 = _ymd8(day or "") @@ -684,8 +1127,8 @@ def build_feed_collect_stats( ) if not heavy: notes.append( - "빠른 조회(기본): 나이컷·호가0종목 교집합은 「상세 집계」로. " - "인덱스(market,code,time)는 종목단위 조회용이라 당일 전수 COUNT/STR_TO_DATE는 여전히 무겁다." + "기본 조회에 폴백나이(usable) 포함. " + "호가0종목 교집합·filter_eval 합격/거부는 「상세 집계」." ) for cov in coverage: if int(cov.get("tick_codes") or 0) <= 0 and int(cov.get("ob_n") or 0) <= 0: @@ -729,19 +1172,9 @@ def build_feed_collect_stats( else: notes.append("LS_WS_TICK_SAVE=true → 구독 종목 US3를 ls_ws_ticks 에 적재 (영구만이 아님).") - if heavy: - age_cut = _feed_age_cut_stats(db, day8, fb_age) - else: - age_cut = { - "age_sec": fb_age, - "empty_minutes_all_fail": None, - "rows": [], - "deferred": True, - "note": ( - f"나이={fb_age:g}s · 기본 조회에서는 생략(전수 STR_TO_DATE 스캔). " - "「상세 집계」로 로드." - ), - } + # 폴백나이: 기본 조회에도 포함 (실측 ~10초대, 요약테이블/크론 불필요) + age_cut = _feed_age_cut_stats(db, day8, fb_age) + disconnect = _feed_disconnect_stats(db, day8) return { "day": f"{day8[:4]}-{day8[4:6]}-{day8[6:8]}", @@ -753,6 +1186,7 @@ def build_feed_collect_stats( "recent_days": [], "recent_days_n": 0, "age_cut": age_cut, + "disconnect": disconnect, "summary": { "ticks": { "kis": tick_kis,