#!/usr/bin/env python3 """Live-ish market timing engine for the account growth agent. Uses yfinance public data. Broker quotes and broker FX must override this for actual order placement. """ from __future__ import annotations import argparse import csv import json import math from dataclasses import dataclass, asdict from datetime import datetime, timezone from pathlib import Path DEFAULT_TICKERS = [ "SPY", "QQQ", "IWM", "SMH", "SOXX", "SOXL", "SOXS", "NVDA", "AVGO", "AMD", "MU", "TSM", "ASML", "RDW", "MRVL", "AAOI", "GNTA", ] LEVERAGED = {"SOXL", "SOXS", "TQQQ", "SQQQ", "UPRO", "SPXL", "SPXS", "QLD", "SSO", "TECL", "TECS"} TREND_KO = { "UPTREND": "상승 추세", "REPAIR": "추세 회복", "PULLBACK": "조정", "DAMAGED": "추세 손상", "TREND BREAK": "추세 이탈", "INTRADAY WEAK": "장중 약세", "CHOP": "혼조", "DATA NEEDED": "데이터 필요", "ERROR": "오류", } TIMING_KO = { "BUY SETUP": "매수 가능", "BUY SETUP - DO NOT CHASE": "매수 후보지만 추격 금지", "REBOUND WATCH": "반등 감시", "INVERSE BUY SETUP": "인버스 매수 후보", "TRIM WATCH": "일부 축소 감시", "EXIT WATCH": "이탈/청산 감시", "NO EDGE": "우위 없음", "DATA NEEDED": "데이터 필요", } ACTION_KO = { "ATTACK": "공격", "GUARDED ATTACK": "제한적 공격", "DEFENSE": "방어", "DEFEND": "방어", "HARVEST": "수익실현", "HARVEST / GUARD": "수익실현/방어", "REST / WAIT": "휴식/대기", "WAIT": "대기", } @dataclass class TimingRow: ticker: str price: float | None prev_close: float | None day_change_pct: float | None sma20: float | None sma50: float | None sma200: float | None rsi14: float | None atr14_pct: float | None volume_ratio: float | None vwap_intraday: float | None vwap_gap_pct: float | None trend_label: str timing_label: str action_label: str entry_trigger: str stop_or_invalidation: str target_1: str target_2: str notes: str def last_valid(series): clean = series.dropna() if clean.empty: return None return float(clean.iloc[-1]) def rsi(close, window=14): delta = close.diff() gain = delta.clip(lower=0).rolling(window).mean() loss = (-delta.clip(upper=0)).rolling(window).mean() rs = gain / loss.replace(0, math.nan) return 100 - (100 / (1 + rs)) def atr_pct(df, window=14): high = df["High"] low = df["Low"] close = df["Close"] prev_close = close.shift(1) tr = (high - low).to_frame("hl") tr["hc"] = (high - prev_close).abs() tr["lc"] = (low - prev_close).abs() atr = tr.max(axis=1).rolling(window).mean() return atr / close def safe_round(value, digits=2): if value is None or value != value: return None return round(float(value), digits) def format_price(value): if value is None or value != value: return "NEEDS LIVE PRICE" return f"{value:.2f}" def row_to_dict(row: TimingRow): data = asdict(row) data["trend_label_ko"] = TREND_KO.get(row.trend_label, row.trend_label) data["timing_label_ko"] = TIMING_KO.get(row.timing_label, row.timing_label) data["action_label_ko"] = ACTION_KO.get(row.action_label, row.action_label) return data def classify_trend(price, sma20, sma50, sma200, rsi_value, vwap_gap): if price is None or sma20 is None or sma50 is None: return "DATA NEEDED" above20 = price > sma20 above50 = price > sma50 above200 = sma200 is not None and price > sma200 if above20 and above50 and above200 and (rsi_value or 50) < 72: return "UPTREND" if above20 and above50: return "REPAIR" if not above20 and above50: return "PULLBACK" if not above50 and sma200 is not None and price > sma200: return "DAMAGED" if sma200 is not None and price < sma200: return "TREND BREAK" if vwap_gap is not None and vwap_gap < -1.5: return "INTRADAY WEAK" return "CHOP" def classify_timing(ticker, price, prev_close, sma20, sma50, sma200, rsi_value, atr_value, volume_ratio, vwap_gap, trend_label): if price is None: return ( "DATA NEEDED", "WAIT", "실시간 가격 확인 필요", "NEEDS LIVE PRICE", "NEEDS LIVE PRICE", "NEEDS LIVE PRICE", "실시간 가격 없이는 행동하지 않습니다.", ) day_change = ((price / prev_close - 1) * 100) if prev_close else None stretched = rsi_value is not None and rsi_value >= 72 washed = rsi_value is not None and rsi_value <= 34 high_volume = volume_ratio is not None and volume_ratio >= 1.5 above_vwap = vwap_gap is not None and vwap_gap >= 0 below_vwap = vwap_gap is not None and vwap_gap <= -0.6 atr = atr_value or 0.04 stop = price * (1 - max(0.025, min(0.10, atr * 1.2))) target_1 = price * (1 + max(0.04, min(0.12, atr * 1.5))) target_2 = price * (1 + max(0.08, min(0.24, atr * 2.8))) if trend_label in {"UPTREND", "REPAIR"} and above_vwap and not stretched: if day_change is not None and day_change > 5 and high_volume: return ("BUY SETUP - DO NOT CHASE", "WAIT", "눌림 또는 VWAP 지지 확인", f"{stop:.2f}", f"{target_1:.2f}", f"{target_2:.2f}", "강하지만 단기 과열입니다. FOMO 진입 금지.") return ("BUY SETUP", "ATTACK", "주요 추세선과 VWAP 위", f"{stop:.2f}", f"{target_1:.2f}", f"{target_2:.2f}", "분할 진입만 검토합니다. 수량은 계좌 모드에 따릅니다.") if trend_label == "PULLBACK" and not below_vwap and washed: return ("REBOUND WATCH", "WAIT", "SMA20 또는 장중 VWAP 회복", f"{stop:.2f}", f"{target_1:.2f}", f"{target_2:.2f}", "반등 가능성은 있지만 아직 확정은 아닙니다.") if trend_label in {"DAMAGED", "TREND BREAK"}: if ticker in {"SOXS", "SQQQ", "SPXS", "TECS"} and above_vwap: return ("INVERSE BUY SETUP", "DEFEND", "하락 확인 및 인버스 VWAP 위", f"{stop:.2f}", f"{target_1:.2f}", f"{target_2:.2f}", "본계좌 전용입니다. 시간 손절을 반드시 정합니다.") return ("EXIT WATCH", "DEFEND", "지지선 또는 추세 이탈", f"{stop:.2f}", f"{target_1:.2f}", f"{target_2:.2f}", "추세 회복 전까지 물타기 금지.") if stretched and high_volume: return ("TRIM WATCH", "HARVEST", "RSI/거래량 과열", f"{stop:.2f}", f"{target_1:.2f}", f"{target_2:.2f}", "수익 잠금 또는 리셋 대기.") return ("NO EDGE", "WAIT", "명확한 트리거 없음", f"{stop:.2f}", f"{target_1:.2f}", f"{target_2:.2f}", "현금도 포지션입니다.") def fetch_one(ticker): import yfinance as yf daily = yf.download(ticker, period="1y", interval="1d", auto_adjust=True, progress=False, threads=False) intraday = yf.download(ticker, period="5d", interval="5m", auto_adjust=True, progress=False, threads=False) return daily, intraday def flatten_if_needed(df, ticker): if df is None or df.empty: return df if hasattr(df.columns, "nlevels") and df.columns.nlevels > 1: if ticker in df.columns.get_level_values(-1): df = df.xs(ticker, axis=1, level=-1) else: df.columns = [col[0] for col in df.columns] return df def analyze_ticker(ticker): daily, intraday = fetch_one(ticker) daily = flatten_if_needed(daily, ticker) intraday = flatten_if_needed(intraday, ticker) if daily is None or daily.empty or "Close" not in daily: return TimingRow(ticker, None, None, None, None, None, None, None, None, None, None, None, "DATA NEEDED", "DATA NEEDED", "WAIT", "live/daily data unavailable", "NEEDS LIVE PRICE", "NEEDS LIVE PRICE", "NEEDS LIVE PRICE", "No action.") close = daily["Close"].dropna() price = last_valid(close) prev_close = float(close.iloc[-2]) if len(close) >= 2 else None day_change_pct = ((price / prev_close - 1) * 100) if price and prev_close else None sma20 = last_valid(close.rolling(20).mean()) sma50 = last_valid(close.rolling(50).mean()) sma200 = last_valid(close.rolling(200).mean()) rsi14 = last_valid(rsi(close, 14)) atr14 = last_valid(atr_pct(daily, 14)) if {"High", "Low", "Close"}.issubset(daily.columns) else None volume_ratio = None if "Volume" in daily and len(daily["Volume"].dropna()) >= 21: last_volume = float(daily["Volume"].dropna().iloc[-1]) avg_volume = float(daily["Volume"].dropna().iloc[-21:-1].mean()) if avg_volume: volume_ratio = last_volume / avg_volume vwap_intraday = None vwap_gap = None if intraday is not None and not intraday.empty and {"High", "Low", "Close", "Volume"}.issubset(intraday.columns): same_day = intraday.dropna().tail(100) typical = (same_day["High"] + same_day["Low"] + same_day["Close"]) / 3 vol = same_day["Volume"].replace(0, math.nan) denom = vol.sum() if denom and denom == denom: vwap_intraday = float((typical * vol).sum() / denom) if price: vwap_gap = (price / vwap_intraday - 1) * 100 trend_label = classify_trend(price, sma20, sma50, sma200, rsi14, vwap_gap) timing, action, trigger, stop, target1, target2, notes = classify_timing( ticker, price, prev_close, sma20, sma50, sma200, rsi14, atr14, volume_ratio, vwap_gap, trend_label ) if ticker in LEVERAGED: notes = notes + " 본계좌 전용." return TimingRow( ticker=ticker, price=safe_round(price), prev_close=safe_round(prev_close), day_change_pct=safe_round(day_change_pct), sma20=safe_round(sma20), sma50=safe_round(sma50), sma200=safe_round(sma200), rsi14=safe_round(rsi14), atr14_pct=safe_round((atr14 * 100) if atr14 else None), volume_ratio=safe_round(volume_ratio), vwap_intraday=safe_round(vwap_intraday), vwap_gap_pct=safe_round(vwap_gap), trend_label=trend_label, timing_label=timing, action_label=action, entry_trigger=trigger, stop_or_invalidation=stop, target_1=target1, target_2=target2, notes=notes, ) def summarize_market(rows): by_ticker = {row.ticker: row for row in rows} qqq = by_ticker.get("QQQ") smh = by_ticker.get("SMH") or by_ticker.get("SOXX") soxl = by_ticker.get("SOXL") soxs = by_ticker.get("SOXS") attacks = [row for row in rows if row.action_label == "ATTACK"] defends = [row for row in rows if row.action_label == "DEFEND"] harvests = [row for row in rows if row.action_label == "HARVEST"] if qqq and smh and qqq.trend_label in {"UPTREND", "REPAIR"} and smh.trend_label in {"UPTREND", "REPAIR"}: mode = "ATTACK" if attacks else "GUARDED ATTACK" elif defends and soxs and soxs.timing_label == "INVERSE BUY SETUP": mode = "DEFENSE" elif harvests: mode = "HARVEST / GUARD" else: mode = "REST / WAIT" top_actions = sorted(rows, key=lambda row: {"ATTACK": 0, "DEFEND": 1, "HARVEST": 2, "WAIT": 3}.get(row.action_label, 4))[:5] return { "mode": mode, "mode_ko": ACTION_KO.get(mode, mode).replace(" / ", "/"), "top_actions": [row_to_dict(row) for row in top_actions], "soxl_status": row_to_dict(soxl) if soxl else None, "soxs_status": row_to_dict(soxs) if soxs else None, } def write_outputs(rows, summary, out_dir: Path): csv_path = out_dir / "live_timing_snapshot.csv" json_path = out_dir / "live_timing_snapshot.json" md_path = out_dir / "live_timing_snapshot.md" with csv_path.open("w", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=list(row_to_dict(rows[0]).keys())) writer.writeheader() writer.writerows([row_to_dict(row) for row in rows]) payload = { "timestamp_utc": datetime.now(timezone.utc).isoformat(), "source": "yfinance public quote/history; broker quote must override for orders", "summary": summary, "rows": [row_to_dict(row) for row in rows], } json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2)) lines = [ "# 실시간 타이밍 스냅샷", "", f"- Timestamp UTC: {payload['timestamp_utc']}", "- Source: yfinance public quote/history; broker quote must override for actual orders.", f"- 현재 모드: `{summary.get('mode_ko', summary['mode'])}` (`{summary['mode']}`)", "", "## 우선 확인", "", "| 티커 | 가격 | 추세 | 타이밍 | 행동 | 조건 | 손절/무효화 | 목표1 | 목표2 |", "| --- | ---: | --- | --- | --- | --- | ---: | ---: | ---: |", ] for row in summary["top_actions"]: lines.append( f"| {row['ticker']} | {format_price(row['price'])} | {row['trend_label_ko']} | {row['timing_label_ko']} | {row['action_label_ko']} | {row['entry_trigger']} | {row['stop_or_invalidation']} | {row['target_1']} | {row['target_2']} |" ) lines.extend([ "", "## 규칙", "", "- `공격(ATTACK)`: 매수 후보가 있어도 계좌 모드, 현금, 목표가, 손절, 수량 확인이 필요합니다.", "- `방어(DEFEND)`: 노출 축소 또는 인버스 검토. 본계좌 적격성과 손절/시간제한이 필요합니다.", "- `수익실현(HARVEST)`: 일부 축소, 수익 잠금, 과열 감시입니다.", "- `대기(WAIT)`: 우위가 부족합니다. 현금도 포지션입니다.", "", "## 파일", "", f"- CSV: `{csv_path}`", f"- JSON: `{json_path}`", ]) md_path.write_text("\n".join(lines) + "\n") def main(): parser = argparse.ArgumentParser() parser.add_argument("--tickers", nargs="*", default=DEFAULT_TICKERS) parser.add_argument("--json", action="store_true") args = parser.parse_args() rows = [] for ticker in args.tickers: try: rows.append(analyze_ticker(ticker.upper())) except Exception as exc: rows.append(TimingRow(ticker.upper(), None, None, None, None, None, None, None, None, None, None, None, "ERROR", "DATA NEEDED", "WAIT", str(exc), "NEEDS LIVE PRICE", "NEEDS LIVE PRICE", "NEEDS LIVE PRICE", "No action.")) summary = summarize_market(rows) out_dir = Path(__file__).resolve().parent write_outputs(rows, summary, out_dir) payload = {"summary": summary, "rows": [row_to_dict(row) for row in rows]} if args.json: print(json.dumps(payload, ensure_ascii=False, indent=2)) else: print(f"현재모드={summary.get('mode_ko', summary['mode'])} ({summary['mode']})") for row in summary["top_actions"]: print(f"{row['ticker']}: {row['timing_label_ko']}({row['timing_label']}) / {row['action_label_ko']}({row['action_label']}) / 가격={format_price(row['price'])} / 손절={row['stop_or_invalidation']} / 목표1={row['target_1']} / 목표2={row['target_2']}") if __name__ == "__main__": main()