예제 소스
실시간 XGBoost와 DQN 결합 구조: 거래 AI를 검증하는 법 · 파일 59개
check_redis_cul_df.py
#!/usr/bin/env python3
"""
Redis `cul_df` 적재 상태 점검 (test_hf_img / ChartImageGenerator 호환).
실행 예:
.venv/bin/python check_redis_cul_df.py
.venv/bin/python check_redis_cul_df.py --loop
.venv/bin/python check_redis_cul_df.py --loop --interval-sec 300
5분 루프(--loop)에서는 직전 스냅샷과 마지막 행 time/date·close·전체 blob 해시를 비교해
갱신 여부를 표시합니다.
종료 코드: 0=정상, 1=실패(연결/키 없음/파싱), 2=redis 미설치
(루프 모드에서 --fail-on-stale 시 데이터가 한 번이라도 멈추면 1로 종료)
"""
from __future__ import annotations
import argparse
import hashlib
import json
import sys
import time
from datetime import datetime
from io import StringIO
from typing import Any
try:
import redis
except ImportError:
print("FAIL: pip install redis 필요", file=sys.stderr)
sys.exit(2)
import pandas as pd
CHART_COLS = [
"open",
"high",
"low",
"close",
"upper",
"ma20",
"ma60",
"ma120",
"lower",
"ma20_deg",
"volume",
"PB",
"MFI10",
]
# 마지막 행 시각 판별 시 우선순위 (time 은 0·오프셋만 있는 경우가 있어 뒤로)
_TIME_COLUMNS = (
"time_stamp",
"timestamp",
"Timestamp",
"date",
"Date",
"datetime",
"time",
)
def _parse_datetime_cell(v: Any) -> tuple[str | None, str | None]:
"""(iso 문자열 또는 원문, 실패/참고 메모). 무효·0 이면 (None, reason)."""
if isinstance(v, bool):
return None, "bool"
if v is None:
return None, "null"
if isinstance(v, float) and pd.isna(v):
return None, "nan"
if isinstance(v, str):
s = v.strip()
if not s or s in ("0", "null", "None"):
return None, "empty_or_zero_str"
# 숫자 문자열(Unix ms 등)
try:
xf = float(s)
if xf == 0.0:
return None, "zero_str"
if not (xf != xf): # not NaN
return _numeric_epoch_to_iso(xf)
except ValueError:
pass
if isinstance(v, (int, float)) and not isinstance(v, bool):
if int(v) == 0 and float(v) == 0.0:
return None, "zero"
return _numeric_epoch_to_iso(float(v))
ts = pd.to_datetime(v, errors="coerce")
if pd.notna(ts):
# 1970 근처만 나오는 잘못된 단위는 아래에서 숫자 재시도 없음(문자열 날짜는 신뢰)
if ts.year >= 1990 or (ts.year == 1970 and isinstance(v, str)):
return ts.isoformat(), None
return None, f"reject_year_{ts.year}"
return (str(v), "unparsed_string")
def _numeric_epoch_to_iso(x: float) -> tuple[str | None, str | None]:
"""Unix 숫자(ms/ns/s 등)를 여러 unit으로 시도해 합리적인 연도가 나오면 채택."""
ax = abs(x)
if ax < 1e-6:
return None, "near_zero"
units_to_try: list[str] = []
if ax >= 1e17:
units_to_try.append("ns")
if ax >= 1e11:
units_to_try.append("ms")
if ax >= 1e9:
units_to_try.append("s")
if not units_to_try:
units_to_try = ["ms", "ns", "s"]
ordered: list[str] = []
_seen: set[str] = set()
for u in units_to_try + ["ms", "ns", "s"]:
if u not in _seen:
_seen.add(u)
ordered.append(u)
for unit in ordered:
try:
ts = pd.to_datetime(x, unit=unit)
y = ts.year
if 1990 <= y <= 2040:
return ts.isoformat(), f"unit={unit}"
except (ValueError, OSError, pd.errors.OutOfBoundsDatetime):
continue
# 마지막으로 unit 없이 (pandas 추론)
try:
ts = pd.to_datetime(x, errors="coerce")
if pd.notna(ts) and 1990 <= ts.year <= 2040:
return ts.isoformat(), "infer"
except Exception:
pass
return None, "no_valid_epoch_unit"
def _snapshot(df: pd.DataFrame, blob: str) -> dict[str, Any]:
"""갱신 여부 판단용 스냅샷 (마지막 행 시각·종가·전체 해시)."""
h = hashlib.sha256(blob.encode("utf-8")).hexdigest()[:20]
last = df.iloc[-1]
time_col: str | None = None
time_last: Any = None
time_raw: Any = None
time_note: str | None = None
for c in _TIME_COLUMNS:
if c not in df.columns:
continue
v = last[c]
time_raw = v
iso, note = _parse_datetime_cell(v)
if iso is not None:
time_col = c
time_last = iso
time_note = note
break
time_note = f"{c}:{note}"
# 디버그: 마지막 행의 시간 관련 원본 (0인지 확인용)
time_debug: dict[str, Any] = {}
for c in _TIME_COLUMNS:
if c in df.columns:
time_debug[c] = last[c]
close_last = None
if "close" in df.columns:
try:
close_last = float(pd.to_numeric(last["close"], errors="coerce"))
except (TypeError, ValueError):
close_last = None
return {
"time_col": time_col,
"time_last": time_last,
"time_raw_last": time_raw,
"time_note": time_note,
"time_debug_last_row": time_debug,
"close_last": close_last,
"blob_hash20": h,
"rows": int(len(df)),
"bytes": len(blob),
}
def _fetch_df(r: redis.Redis, key: str) -> tuple[pd.DataFrame, str]:
raw = r.get(key)
if raw is None:
raise KeyError(key)
blob = raw.decode("utf-8") if isinstance(raw, (bytes, bytearray)) else str(raw)
df = pd.read_json(StringIO(blob), orient="records")
return df, blob
def run_once(args: argparse.Namespace) -> tuple[int, dict]:
"""단일 점검. 반환: (exit_code, out_dict)"""
out: dict = {"host": args.host, "port": args.port, "db": args.db, "key": args.key}
try:
r = redis.StrictRedis(host=args.host, port=args.port, db=args.db, socket_connect_timeout=5)
r.ping()
out["ping"] = "ok"
except Exception as e:
out["ok"] = False
out["error"] = f"redis_connect: {e}"
return 1, out
try:
df, blob = _fetch_df(r, args.key)
except KeyError:
out["ok"] = False
out["error"] = "key_missing"
return 1, out
except Exception as e:
out["ok"] = False
out["error"] = f"read_parse: {e}"
return 1, out
if df.empty:
out["ok"] = False
out["error"] = "empty_dataframe"
return 1, out
out["ok"] = True
out["bytes"] = len(blob)
out["rows"] = int(len(df))
out["cols"] = int(len(df.columns))
out["column_names"] = list(df.columns)
missing = [c for c in CHART_COLS if c not in df.columns]
out["chart_missing_columns"] = missing
out["chart_ready"] = len(missing) == 0
if "close" in df.columns:
s = pd.to_numeric(df["close"], errors="coerce")
out["close_min"] = float(s.min()) if s.notna().any() else None
out["close_max"] = float(s.max()) if s.notna().any() else None
out["close_valid"] = int(s.notna().sum())
if "close" in df.columns:
out["close_tail_last3"] = [float(x) for x in df["close"].tail(3).tolist()]
out["snapshot"] = _snapshot(df, blob)
return 0, out
def _emit(out: dict, as_json: bool) -> None:
if as_json:
print(json.dumps(out, ensure_ascii=False, default=str))
return
ok = out.get("ok")
print(f"연결: {out['host']}:{out['port']} db={out['db']} key={out['key']!r}")
if out.get("ping"):
print("ok: PING")
if not ok:
print(f"FAIL: {out.get('error', 'unknown')}", file=sys.stderr)
return
print(f"ok: 값 {out.get('bytes')} bytes")
print(f"ok: 행 {out.get('rows')} 열 {out.get('cols')}")
cols = out.get("column_names") or []
print("컬럼:", cols[:30], ("..." if len(cols) > 30 else ""))
miss = out.get("chart_missing_columns") or []
if miss:
print("WARN: 차트용 컬럼 누락:", miss)
else:
print("ok: 차트용 필수 컬럼(PB·MFI10 포함) 모두 존재")
if out.get("close_min") is not None:
print(
f"close: min={out['close_min']:.6g} max={out['close_max']:.6g} "
f"유효 {out.get('close_valid')}/{out.get('rows')}"
)
if out.get("close_tail_last3"):
print("마지막 3행 close:", out["close_tail_last3"])
snap = out.get("snapshot") or {}
if snap:
print(
f"스냅샷: time_col={snap.get('time_col')!r} "
f"time_last={snap.get('time_last')!r} close_last={snap.get('close_last')} "
f"blob_hash={snap.get('blob_hash20')}"
)
if snap.get("time_note"):
print(f" time_note: {snap.get('time_note')}")
dbg = snap.get("time_debug_last_row") or {}
if dbg:
print(f" 마지막 행 시간 관련 원본: {dbg}")
def _compare_snap(prev: dict | None, cur: dict) -> str:
"""이전 스냅샷 대비 상태 문자열."""
if prev is None:
return "첫 샘플 (비교 없음)"
a, b = prev, cur
if a.get("blob_hash20") != b.get("blob_hash20"):
return "갱신됨 (페이로드 해시 변경)"
if a.get("time_last") != b.get("time_last"):
return "갱신됨 (마지막 시각 변경)"
if a.get("close_last") != b.get("close_last"):
return "갱신됨 (마지막 종가 변경)"
if a.get("rows") != b.get("rows"):
return "갱신됨 (행 수 변경)"
return "경고: 직전 주기와 동일 — 데이터가 안 바뀐 것으로 보임 (스냅샷 동일)"
def run_loop(args: argparse.Namespace) -> int:
prev_snap: dict | None = None
stale_count = 0
exit_code = 0
print(
f"루프 시작: {args.interval_sec}초 간격, 키={args.key!r} "
f"(종료: Ctrl+C)",
file=sys.stderr,
)
try:
iteration = 0
while True:
iteration += 1
now = datetime.now().isoformat(timespec="seconds")
code, out = run_once(args)
snap = (out.get("snapshot") or {}) if out.get("ok") else {}
if args.json:
line = {
"ts_local": now,
"iteration": iteration,
"exit": code,
**{k: v for k, v in out.items() if k != "snapshot"},
"snapshot": snap,
}
if prev_snap is not None and out.get("ok"):
line["change_status"] = _compare_snap(prev_snap, snap)
print(json.dumps(line, ensure_ascii=False, default=str), flush=True)
else:
print(f"\n--- [{now}] 반복 #{iteration} ---", flush=True)
if code != 0:
print(f"FAIL: {out.get('error')}", file=sys.stderr, flush=True)
if args.fail_on_error:
return 1
else:
_emit(out, as_json=False)
if prev_snap is not None:
status = _compare_snap(prev_snap, snap)
print(status, flush=True)
if "경고" in status or "동일" in status:
stale_count += 1
if args.fail_on_stale:
print("FAIL: --fail-on-stale", file=sys.stderr)
return 1
if out.get("ok") and snap:
prev_snap = snap
time.sleep(args.interval_sec)
except KeyboardInterrupt:
print("\n종료 (KeyboardInterrupt)", file=sys.stderr)
return exit_code
def main() -> int:
ap = argparse.ArgumentParser(description="Redis cul_df 키 점검")
ap.add_argument("--host", default="localhost")
ap.add_argument("--port", type=int, default=6379)
ap.add_argument("--db", type=int, default=0)
ap.add_argument("--key", default="cul_df")
ap.add_argument("--json", action="store_true", help="한 줄 JSON으로 출력")
ap.add_argument(
"--strict",
action="store_true",
help="차트용 컬럼(PB·MFI10 등)이 하나라도 없으면 종료 코드 1",
)
ap.add_argument(
"--loop",
action="store_true",
help="주기적으로 반복 점검 (기본 300초=5분)",
)
ap.add_argument(
"--interval-sec",
type=int,
default=300,
metavar="N",
help="루프 간격(초). 기본 300 (5분)",
)
ap.add_argument(
"--fail-on-stale",
action="store_true",
help="루프에서 한 번이라도 스냅샷이 바뀌지 않으면 종료 코드 1",
)
ap.add_argument(
"--fail-on-error",
action="store_true",
help="루프에서 Redis/파싱 오류 시 즉시 종료",
)
args = ap.parse_args()
if args.loop:
if args.interval_sec < 1:
print("FAIL: --interval-sec >= 1", file=sys.stderr)
return 1
return run_loop(args)
code, out = run_once(args)
_emit(out, args.json)
if not out.get("ok"):
return 1
if args.strict and not out.get("chart_ready"):
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())