472 lines
19 KiB
Python
472 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
LS 섀도 시세 유니버스 히스토리 수집 + 종목 히스토리 퀄리티 리포트 (단독 실행).
|
||
|
||
테이블: ``ls_universe_history``
|
||
- target_candidates_history(키움 조건식)와 분리
|
||
- 슬롯(기본 1분)마다 LS가 실제로 틱/봉을 쌓은 종목을 스냅샷
|
||
|
||
사용:
|
||
# 오늘 틱으로 슬롯 백필 + 퀄리티 출력
|
||
python3 scripts/collect_ls_universe_history.py --day 2026-07-27 --backfill --quality
|
||
|
||
# 현재 시각 슬롯 1회 수집만
|
||
python3 scripts/collect_ls_universe_history.py --collect
|
||
|
||
# N초마다 수집 (Ctrl+C 종료)
|
||
python3 scripts/collect_ls_universe_history.py --loop 60 --quality-every 5
|
||
|
||
# 퀄리티만 (이미 쌓인 테이블/틱/봉 기준)
|
||
python3 scripts/collect_ls_universe_history.py --day 2026-07-27 --quality
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import os
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from typing import Any, Dict, List, Optional, Sequence, Tuple
|
||
|
||
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
if _ROOT not in sys.path:
|
||
sys.path.insert(0, _ROOT)
|
||
|
||
|
||
DDL = """
|
||
CREATE TABLE IF NOT EXISTS ls_universe_history (
|
||
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
|
||
slot_key VARCHAR(12) NOT NULL,
|
||
scan_time VARCHAR(30) NOT NULL,
|
||
code VARCHAR(20) NOT NULL,
|
||
name VARCHAR(100) NOT NULL DEFAULT '',
|
||
price DOUBLE NOT NULL DEFAULT 0,
|
||
tick_count INT NOT NULL DEFAULT 0,
|
||
candle_1m_count INT NOT NULL DEFAULT 0,
|
||
last_tick_ts DATETIME(3) NULL,
|
||
last_candle_dt VARCHAR(20) NULL,
|
||
gap_1m_missing INT NOT NULL DEFAULT 0,
|
||
source VARCHAR(16) NOT NULL DEFAULT 'ls_ws',
|
||
note VARCHAR(200) NULL,
|
||
UNIQUE KEY uq_ls_univ_slot_code (slot_key, code),
|
||
INDEX idx_ls_univ_slot (slot_key),
|
||
INDEX idx_ls_univ_code (code),
|
||
INDEX idx_ls_univ_scan (scan_time)
|
||
) CHARACTER SET utf8mb4
|
||
"""
|
||
|
||
|
||
def _parse_day(s: str) -> str:
|
||
s = (s or "").strip()
|
||
if not s:
|
||
return datetime.now().strftime("%Y-%m-%d")
|
||
if len(s) == 8 and s.isdigit():
|
||
return f"{s[:4]}-{s[5:7]}-{s[6:8]}" if False else f"{s[0:4]}-{s[4:6]}-{s[6:8]}"
|
||
return s[:10]
|
||
|
||
|
||
def _day_compact(day: str) -> str:
|
||
return day.replace("-", "")[:8]
|
||
|
||
|
||
def _slot_key_from_dt(dt: datetime, slot_min: int = 1) -> str:
|
||
slot_min = max(1, int(slot_min))
|
||
m = (dt.minute // slot_min) * slot_min
|
||
return dt.strftime("%Y%m%d%H") + f"{m:02d}"
|
||
|
||
|
||
def _ensure_table(db) -> None:
|
||
db.conn.execute(DDL)
|
||
|
||
|
||
def _name_map(db, codes: Sequence[str]) -> Dict[str, str]:
|
||
out: Dict[str, str] = {}
|
||
if not codes:
|
||
return out
|
||
# target_candidates 에 있으면 이름 재사용
|
||
try:
|
||
cols = [r["Field"] for r in db.conn.execute("SHOW COLUMNS FROM target_candidates").fetchall()]
|
||
if "code" in cols and "name" in cols:
|
||
# chunk
|
||
codes = [str(c) for c in codes]
|
||
for i in range(0, len(codes), 200):
|
||
chunk = codes[i : i + 200]
|
||
ph = ",".join(["%s"] * len(chunk))
|
||
rows = db.conn.execute(
|
||
f"SELECT code, name FROM target_candidates WHERE code IN ({ph})",
|
||
tuple(chunk),
|
||
).fetchall()
|
||
for r in rows:
|
||
out[str(r["code"])] = str(r.get("name") or "")
|
||
except Exception:
|
||
pass
|
||
return out
|
||
|
||
|
||
def _market_1m_slots(day: str) -> List[str]:
|
||
"""정규장 09:00~15:29 1분 슬롯 키 목록."""
|
||
d = _day_compact(day)
|
||
out: List[str] = []
|
||
t0 = datetime.strptime(d + "0900", "%Y%m%d%H%M")
|
||
t1 = datetime.strptime(d + "1530", "%Y%m%d%H%M")
|
||
cur = t0
|
||
while cur < t1:
|
||
out.append(cur.strftime("%Y%m%d%H%M"))
|
||
cur += timedelta(minutes=1)
|
||
return out
|
||
|
||
|
||
def collect_slot(db, *, slot_key: Optional[str] = None, slot_min: int = 1) -> int:
|
||
"""현재(또는 지정) 슬롯에 LS 틱이 있는 종목을 ls_universe_history 에 upsert."""
|
||
_ensure_table(db)
|
||
now = datetime.now()
|
||
sk = slot_key or _slot_key_from_dt(now, slot_min)
|
||
# 슬롯 구간
|
||
try:
|
||
slot_dt = datetime.strptime(sk, "%Y%m%d%H%M")
|
||
except ValueError:
|
||
slot_dt = now.replace(second=0, microsecond=0)
|
||
slot_end = slot_dt + timedelta(minutes=max(1, slot_min))
|
||
ts0 = slot_dt.strftime("%Y-%m-%d %H:%M:%S")
|
||
ts1 = slot_end.strftime("%Y-%m-%d %H:%M:%S")
|
||
day = slot_dt.strftime("%Y-%m-%d")
|
||
|
||
rows = db.conn.execute(
|
||
"SELECT a.code AS code, a.tick_count AS tick_count, a.last_tick_ts AS last_tick_ts, "
|
||
"b.price AS last_price "
|
||
"FROM ("
|
||
" SELECT code, COUNT(*) AS tick_count, MAX(ts) AS last_tick_ts "
|
||
" FROM ls_ws_ticks WHERE ts >= %s AND ts < %s "
|
||
" GROUP BY code"
|
||
") a "
|
||
"LEFT JOIN ls_ws_ticks b ON b.code = a.code AND b.ts = a.last_tick_ts",
|
||
(ts0, ts1),
|
||
).fetchall()
|
||
if not rows:
|
||
return 0
|
||
|
||
codes = [str(r["code"]) for r in rows]
|
||
names = _name_map(db, codes)
|
||
scan_time = now.strftime("%Y-%m-%d %H:%M:%S")
|
||
n = 0
|
||
for r in rows:
|
||
code = str(r["code"])
|
||
# 당일 1분봉 수 + 최신 봉
|
||
cinfo = db.conn.execute(
|
||
"SELECT COUNT(*) n, MAX(datetime) mx FROM ls_ws_candles "
|
||
"WHERE code=%s AND tf_min=1 AND datetime LIKE %s",
|
||
(code, f"{day}%"),
|
||
).fetchone()
|
||
c_n = int((cinfo or {}).get("n") or 0)
|
||
c_mx = (cinfo or {}).get("mx")
|
||
# 간단 갭: 장 시작~해당 슬롯까지 기대 1분봉 − 보유
|
||
expect = sum(1 for s in _market_1m_slots(day) if s <= sk)
|
||
gap = max(0, expect - c_n) if expect > 0 else 0
|
||
price = float(r.get("last_price") or 0)
|
||
db.conn.execute(
|
||
"INSERT INTO ls_universe_history "
|
||
"(slot_key, scan_time, code, name, price, tick_count, candle_1m_count, "
|
||
" last_tick_ts, last_candle_dt, gap_1m_missing, source, note) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
|
||
"ON DUPLICATE KEY UPDATE "
|
||
"scan_time=VALUES(scan_time), name=VALUES(name), price=VALUES(price), "
|
||
"tick_count=VALUES(tick_count), candle_1m_count=VALUES(candle_1m_count), "
|
||
"last_tick_ts=VALUES(last_tick_ts), last_candle_dt=VALUES(last_candle_dt), "
|
||
"gap_1m_missing=VALUES(gap_1m_missing), note=VALUES(note)",
|
||
(
|
||
sk,
|
||
scan_time,
|
||
code,
|
||
names.get(code, ""),
|
||
price,
|
||
int(r.get("tick_count") or 0),
|
||
c_n,
|
||
r.get("last_tick_ts"),
|
||
str(c_mx or "")[:20] or None,
|
||
int(gap),
|
||
"ls_ws",
|
||
f"slot_min={slot_min}",
|
||
),
|
||
)
|
||
n += 1
|
||
return n
|
||
|
||
|
||
def backfill_day(db, day: str, *, slot_min: int = 1) -> Tuple[int, int]:
|
||
"""해당 일의 ls_ws_ticks 를 분 슬롯으로 묶어 히스토리 upsert. (slots, rows)"""
|
||
_ensure_table(db)
|
||
d = _day_compact(day)
|
||
ts0 = f"{day} 00:00:00"
|
||
ts1 = f"{day} 23:59:59.999"
|
||
slots = db.conn.execute(
|
||
"SELECT DATE_FORMAT(ts, '%%Y%%m%%d%%H%%i') sk FROM ls_ws_ticks "
|
||
"WHERE ts >= %s AND ts <= %s GROUP BY sk ORDER BY sk",
|
||
(ts0, ts1),
|
||
).fetchall()
|
||
# DATE_FORMAT minute is exact; for slot_min>1 floor in python
|
||
seen = set()
|
||
slot_list: List[str] = []
|
||
for r in slots:
|
||
sk_raw = str(r["sk"])
|
||
if len(sk_raw) < 12:
|
||
continue
|
||
if slot_min <= 1:
|
||
sk = sk_raw[:12]
|
||
else:
|
||
dt = datetime.strptime(sk_raw[:12], "%Y%m%d%H%M")
|
||
sk = _slot_key_from_dt(dt, slot_min)
|
||
if sk not in seen:
|
||
seen.add(sk)
|
||
slot_list.append(sk)
|
||
total_rows = 0
|
||
for sk in slot_list:
|
||
if not sk.startswith(d):
|
||
continue
|
||
total_rows += collect_slot(db, slot_key=sk, slot_min=slot_min)
|
||
return len(slot_list), total_rows
|
||
|
||
|
||
def quality_report(db, day: str, *, top: int = 20, since: str = "00:00:00") -> None:
|
||
"""LS 틱/봉/히스토리/키움후보 대비 퀄리티 출력."""
|
||
_ensure_table(db)
|
||
d = _day_compact(day)
|
||
start_ts = f"{day} {since}"
|
||
end_ts = f"{day} 23:59:59.999"
|
||
s_compact = since.replace(":", "")
|
||
while len(s_compact) < 6:
|
||
s_compact += "0"
|
||
start_dt_str = f"{d}{s_compact[:6]}"
|
||
start_hm_str = f"{d}{s_compact[:4]}"
|
||
|
||
print("=" * 72)
|
||
print(f"LS 종목 히스토리 퀄리티 — day={day} (분석 기준 시각: {since} ~ 23:59:59)")
|
||
print("=" * 72)
|
||
|
||
print(f"\n[1. 실매매 틱 / 캔들 / 호가 적재 현황 — {day} {since} 이후]")
|
||
tick_ls = db.conn.execute(
|
||
"SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(ts) mn, MAX(ts) mx "
|
||
"FROM ls_ws_ticks WHERE ts >= %s AND ts <= %s",
|
||
(start_ts, end_ts),
|
||
).fetchone()
|
||
print(f" 🔸 [LS 틱 (ls_ws_ticks)] n={tick_ls['n']:>9,d} | codes={tick_ls['codes']:>4} | range={tick_ls['mn']} ~ {tick_ls['mx']}")
|
||
|
||
try:
|
||
tick_kis = db.conn.execute(
|
||
"SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(tick_time) mn, MAX(tick_time) mx "
|
||
"FROM ws_ticks WHERE tick_time >= %s AND tick_time <= %s",
|
||
(start_dt_str, f"{d}235959"),
|
||
).fetchone()
|
||
print(f" 🔸 [KIS 틱 (ws_ticks)] n={tick_kis['n']:>9,d} | codes={tick_kis['codes']:>4} | range={tick_kis['mn']} ~ {tick_kis['mx']}")
|
||
except Exception as e:
|
||
print(f" 🔸 [KIS 틱 (ws_ticks)] 조회스킵({e})")
|
||
|
||
cndl_ls = db.conn.execute(
|
||
"SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(datetime) mn, MAX(datetime) mx "
|
||
"FROM ls_ws_candles WHERE tf_min=1 AND datetime >= %s AND datetime <= %s",
|
||
(start_ts, end_ts),
|
||
).fetchone()
|
||
print(f" 🔸 [LS 분봉 (ls_ws_candles)] n={cndl_ls['n']:>9,d} | codes={cndl_ls['codes']:>4} | range={cndl_ls['mn']} ~ {cndl_ls['mx']}")
|
||
|
||
try:
|
||
cndl_kis = db.conn.execute(
|
||
"SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(candle_time) mn, MAX(candle_time) mx "
|
||
"FROM ws_candles WHERE timeframe=1 AND candle_time >= %s AND candle_time <= %s",
|
||
(start_hm_str, f"{d}2359"),
|
||
).fetchone()
|
||
print(f" 🔸 [KIS 분봉 (ws_candles)] n={cndl_kis['n']:>9,d} | codes={cndl_kis['codes']:>4} | range={cndl_kis['mn']} ~ {cndl_kis['mx']}")
|
||
except Exception as e:
|
||
print(f" 🔸 [KIS 분봉 (ws_candles)] 조회스킵({e})")
|
||
|
||
try:
|
||
ob_ls = db.conn.execute(
|
||
"SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(snap_time) mn, MAX(snap_time) mx "
|
||
"FROM ls_ws_orderbook WHERE snap_time >= %s AND snap_time <= %s",
|
||
(start_dt_str, f"{d}235959"),
|
||
).fetchone()
|
||
print(f" 🔸 [LS 호가 (ls_ws_orderbook)] n={ob_ls['n']:>9,d} | codes={ob_ls['codes']:>4} | range={ob_ls['mn']} ~ {ob_ls['mx']}")
|
||
except Exception as e:
|
||
print(f" 🔸 [LS 호가 (ls_ws_orderbook)] 조회스킵({e})")
|
||
|
||
try:
|
||
ob_kis = db.conn.execute(
|
||
"SELECT COUNT(*) n, COUNT(DISTINCT code) codes, MIN(snap_time) mn, MAX(snap_time) mx "
|
||
"FROM ws_orderbook WHERE snap_time >= %s AND snap_time <= %s",
|
||
(start_dt_str, f"{d}235959"),
|
||
).fetchone()
|
||
print(f" 🔸 [KIS 호가 (ws_orderbook)] n={ob_kis['n']:>9,d} | codes={ob_kis['codes']:>4} | range={ob_kis['mn']} ~ {ob_kis['mx']}")
|
||
except Exception as e:
|
||
print(f" 🔸 [KIS 호가 (ws_orderbook)] 조회스킵({e})")
|
||
|
||
hist = db.conn.execute(
|
||
"SELECT COUNT(*) n, COUNT(DISTINCT code) codes, COUNT(DISTINCT slot_key) slots "
|
||
"FROM ls_universe_history WHERE slot_key LIKE %s",
|
||
(f"{d}%",),
|
||
).fetchone()
|
||
print(
|
||
f"[ls_universe_history] rows={hist['n']:,} codes={hist['codes']} "
|
||
f"slots={hist['slots']}"
|
||
)
|
||
|
||
# 종목별 틱/봉
|
||
per = db.conn.execute(
|
||
"SELECT code, COUNT(*) ticks, MIN(ts) mn, MAX(ts) mx "
|
||
"FROM ls_ws_ticks WHERE ts >= %s AND ts <= %s "
|
||
"GROUP BY code ORDER BY ticks DESC LIMIT %s",
|
||
(f"{day} 00:00:00", f"{day} 23:59:59.999", int(top)),
|
||
).fetchall()
|
||
print(f"\n--- 틱 상위 {top} 종목 ---")
|
||
print(f"{'code':8} {'ticks':>8} {'candle_1m':>9} {'gap_est':>7} {'last_tick'}")
|
||
expect_full = len(_market_1m_slots(day))
|
||
now_sk = _slot_key_from_dt(datetime.now(), 1)
|
||
if now_sk.startswith(d):
|
||
expect_now = sum(1 for s in _market_1m_slots(day) if s <= now_sk)
|
||
else:
|
||
expect_now = expect_full
|
||
for r in per:
|
||
code = str(r["code"])
|
||
cn = db.conn.execute(
|
||
"SELECT COUNT(*) n FROM ls_ws_candles WHERE code=%s AND tf_min=1 AND datetime LIKE %s",
|
||
(code, f"{day}%"),
|
||
).fetchone()["n"]
|
||
gap = max(0, int(expect_now) - int(cn))
|
||
print(
|
||
f"{code:8} {int(r['ticks']):8d} {int(cn):9d} {gap:7d} {r['mx']}"
|
||
)
|
||
|
||
# 키움 후보 히스토리와 교집합 (오늘)
|
||
try:
|
||
kw = db.conn.execute(
|
||
"SELECT COUNT(DISTINCT code) n FROM target_candidates_history "
|
||
"WHERE slot_key LIKE %s",
|
||
(f"{d}%",),
|
||
).fetchone()["n"]
|
||
both = db.conn.execute(
|
||
"SELECT COUNT(DISTINCT h.code) n FROM ls_universe_history h "
|
||
"INNER JOIN target_candidates_history t "
|
||
" ON t.code=h.code AND t.slot_key LIKE %s "
|
||
"WHERE h.slot_key LIKE %s",
|
||
(f"{d}%", f"{d}%"),
|
||
).fetchone()["n"]
|
||
ls_only = db.conn.execute(
|
||
"SELECT COUNT(DISTINCT code) n FROM ls_universe_history WHERE slot_key LIKE %s",
|
||
(f"{d}%",),
|
||
).fetchone()["n"]
|
||
print(f"\n--- 키움 후보(history) vs LS 히스토리 ---")
|
||
print(f"키움 distinct codes(today): {kw}")
|
||
print(f"LS history distinct: {ls_only}")
|
||
print(f"교집합(대략): {both}")
|
||
except Exception as e:
|
||
print(f"(키움 교차 비교 스킵: {e})")
|
||
|
||
print("\n[2. 증권사 간 시세 및 레이텐시(수신 지연) 3자 비교 실측]")
|
||
def _fmt(val):
|
||
return f"{val:.1f}ms" if val is not None else "N/A"
|
||
|
||
try:
|
||
v1 = db.conn.execute(
|
||
"SELECT COUNT(*) n, AVG(ABS(diff_pct)) avg_abs, MAX(ABS(diff_pct)) max_abs, "
|
||
"AVG(kis_age_ms) kis_ms, AVG(kiwoom_age_ms) kw_ms "
|
||
"FROM ws_price_validation WHERE ts >= %s AND ts <= %s AND diff_pct IS NOT NULL",
|
||
(start_ts, end_ts),
|
||
).fetchone()
|
||
if v1 and v1["n"] > 0:
|
||
print(f" 1) KIS ↔ 키움 시세 갭 실측 (ws_price_validation, 표본: {v1['n']:,}건)")
|
||
print(f" 👉 가격 괴리율: 평균 {v1['avg_abs']:.4f}% (최대 {v1['max_abs']:.4f}%)")
|
||
print(f" 👉 수신 지연(Age): KIS 평균 {_fmt(v1['kis_ms'])} vs 키움 평균 {_fmt(v1['kw_ms'])}")
|
||
else:
|
||
print(" 1) KIS ↔ 키움 (ws_price_validation): 당일 실측 표본 없음")
|
||
except Exception as e:
|
||
print(f" (ws_price_validation 스킵: {e})")
|
||
|
||
try:
|
||
v2 = db.conn.execute(
|
||
"SELECT COUNT(*) n, "
|
||
"AVG(ABS(diff_kis_ls_pct)) kis_ls_avg, MAX(ABS(diff_kis_ls_pct)) kis_ls_max, "
|
||
"AVG(ABS(diff_kw_ls_pct)) kw_ls_avg, MAX(ABS(diff_kw_ls_pct)) kw_ls_max, "
|
||
"AVG(kis_age_ms) kis_ms, AVG(kiwoom_age_ms) kw_ms, AVG(ls_age_ms) ls_ms "
|
||
"FROM ws_price_validation_ls "
|
||
"WHERE ts >= %s AND ts <= %s AND diff_kis_ls_pct IS NOT NULL",
|
||
(start_ts, end_ts),
|
||
).fetchone()
|
||
if v2 and v2["n"] > 0:
|
||
print(f" 2) KIS ↔ 키움 ↔ LS 삼각 시세 갭 실측 (ws_price_validation_ls, 표본: {v2['n']:,}건)")
|
||
print(f" 👉 KIS vs LS 가격 괴리율: 평균 {v2['kis_ls_avg']:.4f}% (최대 {v2['kis_ls_max']:.4f}%)")
|
||
print(f" 👉 키움 vs LS 가격 괴리율: 평균 {v2['kw_ls_avg']:.4f}% (최대 {v2['kw_ls_max']:.4f}%)")
|
||
print(f" 👉 수신 지연(Age): KIS 평균 {_fmt(v2['kis_ms'])} vs 키움 평균 {_fmt(v2['kw_ms'])} vs LS 평균 {_fmt(v2['ls_ms'])}")
|
||
else:
|
||
print(" 2) KIS ↔ 키움 ↔ LS (ws_price_validation_ls): 당일 실측 표본 없음")
|
||
except Exception as e:
|
||
print(f" (ws_price_validation_ls 스킵: {e})")
|
||
|
||
# 최근 슬롯 커버
|
||
recent = db.conn.execute(
|
||
"SELECT slot_key, COUNT(*) n FROM ls_universe_history "
|
||
"WHERE slot_key LIKE %s GROUP BY slot_key ORDER BY slot_key DESC LIMIT 8",
|
||
(f"{d}%",),
|
||
).fetchall()
|
||
print("\n--- 최근 슬롯 종목수 ---")
|
||
for r in recent:
|
||
print(f" {r['slot_key']}: {r['n']} codes")
|
||
print("=" * 72)
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="LS universe history collect + quality")
|
||
ap.add_argument("--day", default="", help="YYYY-MM-DD (기본=오늘)")
|
||
ap.add_argument("--slot-min", type=int, default=1, help="슬롯 분 (기본 1)")
|
||
ap.add_argument("--collect", action="store_true", help="현재 슬롯 1회 수집")
|
||
ap.add_argument("--backfill", action="store_true", help="해당일 틱→슬롯 백필")
|
||
ap.add_argument("--quality", action="store_true", help="퀄리티 리포트 출력")
|
||
ap.add_argument("--loop", type=int, default=0, help="N초마다 --collect 반복 (0=OFF)")
|
||
ap.add_argument("--quality-every", type=int, default=0, help="루프 N회마다 quality")
|
||
ap.add_argument("--top", type=int, default=20, help="퀄리티 상위 종목 수")
|
||
ap.add_argument("--since", default="00:00:00", help="이 시각(HH:MM:SS) 이후부터 분석 (기본 00:00:00)")
|
||
args = ap.parse_args()
|
||
|
||
day = _parse_day(args.day)
|
||
from database import TradeDB
|
||
|
||
db = TradeDB()
|
||
try:
|
||
_ensure_table(db)
|
||
print(f"[ok] ls_universe_history ready | day={day} slot_min={args.slot_min}")
|
||
|
||
if args.backfill:
|
||
ns, nr = backfill_day(db, day, slot_min=max(1, args.slot_min))
|
||
print(f"[backfill] slots={ns} upsert_rows≈{nr}")
|
||
|
||
if args.collect and not args.loop:
|
||
n = collect_slot(db, slot_min=max(1, args.slot_min))
|
||
print(f"[collect] upsert codes={n} slot={_slot_key_from_dt(datetime.now(), args.slot_min)}")
|
||
|
||
if args.quality and not args.loop:
|
||
quality_report(db, day, top=max(5, args.top), since=args.since)
|
||
|
||
if args.loop and args.loop > 0:
|
||
i = 0
|
||
print(f"[loop] every {args.loop}s — Ctrl+C to stop")
|
||
while True:
|
||
n = collect_slot(db, slot_min=max(1, args.slot_min))
|
||
sk = _slot_key_from_dt(datetime.now(), args.slot_min)
|
||
print(f"[{datetime.now():%H:%M:%S}] collect slot={sk} codes={n}")
|
||
i += 1
|
||
if args.quality_every > 0 and i % args.quality_every == 0:
|
||
quality_report(db, day, top=max(5, args.top), since=args.since)
|
||
time.sleep(max(5, args.loop))
|
||
|
||
# 아무 플래그 없으면 backfill+quality 기본
|
||
if not any([args.collect, args.backfill, args.quality, args.loop]):
|
||
ns, nr = backfill_day(db, day, slot_min=max(1, args.slot_min))
|
||
print(f"[backfill] slots={ns} upsert_rows≈{nr}")
|
||
quality_report(db, day, top=max(5, args.top), since=args.since)
|
||
finally:
|
||
try:
|
||
db.close()
|
||
except Exception:
|
||
pass
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|