#!/usr/bin/env python3 """보유종목 현재가 주기 갱신 (Yahoo Finance / yfinance 소스). 매매(수량·평단) 변화가 없을 때, live_account_state.json/.js 의 보유종목 current_price·value_krw·pnl_krw 만 최신 시세로 갱신한다. 엄격 규칙(이 프로젝트): - 보유 수량(qty)·원가(buy basis)는 절대 수정하지 않는다. (매매 시에만 사용자 입력) - 전략/판단 필드(mode, verdict, kakao_summary 등)도 건드리지 않는다. - 갱신 대상은 orders[].current_price / value_krw / pnl_krw 와 메타(prices_updated_at, fx_usdkrw) 뿐. - 데이터 실패 종목은 기존 값 유지(임의 0/추정 금지). 무료 Yahoo 소스만 사용. """ from __future__ import annotations import csv as csvmod import datetime as dt import glob import json import os import urllib.request BASE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(BASE) ENV_FILE = os.path.join(ROOT, ".kakao_env") JSON_PATH = os.path.join(BASE, "live_account_state.json") JS_PATH = os.path.join(BASE, "live_account_state.js") LOG_PATH = os.path.join(BASE, "price_refresh_log.jsonl") JS_PREFIX = "window.LIVE_ACCOUNT_STATE = " UA = {"User-Agent": "Mozilla/5.0"} def yahoo_price(symbol: str): """확장시간(프리/애프터마켓) 포함 최신가. 1분봉 includePrePost=true 의 마지막 유효 종가를 우선 사용하고, post/pre meta가 있으면 그 값을, 없으면 정규장가로 폴백.""" url = (f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}" f"?range=1d&interval=1m&includePrePost=true") req = urllib.request.Request(url, headers=UA) res = json.loads(urllib.request.urlopen(req, timeout=15).read())["chart"]["result"][0] meta = res.get("meta", {}) # 1) 확장시간 포함 1분봉 마지막 유효 종가 (가장 최신 체결가) try: closes = res["indicators"]["quote"][0]["close"] for v in reversed(closes): if v is not None: return float(v) except Exception: pass # 2) meta의 post/pre/정규 순 폴백 for k in ("postMarketPrice", "preMarketPrice", "regularMarketPrice"): if meta.get(k) is not None: return float(meta[k]) return None def read_fx_spread() -> float: """증권사 기준환율 보정 계수. .kakao_env 의 FX_SPREAD(예: 1.003) 또는 env. 기본 1.0(=시장 스팟 그대로). effective_fx = 시장스팟 × FX_SPREAD.""" v = os.environ.get("FX_SPREAD") if not v and os.path.exists(ENV_FILE): try: for line in open(ENV_FILE, encoding="utf-8"): if line.strip().startswith("FX_SPREAD="): v = line.strip().split("=", 1)[1] break except Exception: pass try: s = float(v) if 0.9 <= s <= 1.1: # 비정상 값 방어 return s except (TypeError, ValueError): pass return 1.0 def read_fx_override() -> float | None: """증권사 기준환율 고정값. .kakao_env 의 FX_RATE(예: 1528.7) 또는 env. 설정 시 평가에 이 값을 그대로 적용(앱과 정확히 일치). 미설정이면 None(=시장 스팟 사용).""" v = os.environ.get("FX_RATE") if not v and os.path.exists(ENV_FILE): try: for line in open(ENV_FILE, encoding="utf-8"): if line.strip().startswith("FX_RATE="): v = line.strip().split("=", 1)[1] break except Exception: pass try: r = float(v) if 800 <= r <= 2500: # 비정상 값 방어 return r except (TypeError, ValueError): pass return None def log(rec: dict) -> None: rec["ts"] = dt.datetime.now(dt.timezone.utc).isoformat() try: with open(LOG_PATH, "a", encoding="utf-8") as f: f.write(json.dumps(rec, ensure_ascii=False) + "\n") except Exception: pass def latest_positions_csv() -> str | None: files = sorted(glob.glob(os.path.join(BASE, "current_positions_from_screenshots_*.csv"))) return files[-1] if files else None def update_accounts(state: dict, prices: dict, fx) -> dict: """계좌별(본/부) 산출을 라이브 시세로 갱신. 보유는 current_positions_from_screenshots_*.csv(매매 시에만 변경)를 기준, 현금은 스냅샷 총액 - 스냅샷 포지션합으로 1회 고정 → current = cash + Σ(live qty*price*fx).""" path = latest_positions_csv() if not path or not fx: return {"ok": False, "reason": "no_csv_or_fx"} rows = {} with open(path, encoding="utf-8") as f: for r in csvmod.DictReader(f): rows.setdefault(r["account"], []).append(r) result = {} for acct, rs in rows.items(): acc = state.get("accounts", {}).get(acct) if not acc: continue pos_base = sum(float(r.get("value_krw") or 0) for r in rs) # 스냅샷 포지션합(고정) # 현금: 최초 1회 스냅샷에서 고정 (이후 보유 CSV가 같으면 동일) if "cash_krw" not in acc: acc["snapshot_total_krw"] = acc.get("current_krw") acc["cash_krw"] = round((acc.get("current_krw") or pos_base) - pos_base) live_val = 0.0 ok = True for r in rs: p = prices.get(r["ticker"]) if p is None: # 시세 실패분은 스냅샷 평가액 사용(왜곡 방지) live_val += float(r.get("value_krw") or 0); ok = False else: live_val += float(r["qty"]) * float(p) * float(fx) acc["positions_value_krw"] = round(live_val) acc["current_krw"] = round(acc.get("cash_krw", 0) + live_val) result[acct] = {"current_krw": acc["current_krw"], "all_priced": ok} return {"ok": True, "accounts": result, "csv": os.path.basename(path)} def atomic_write(path: str, text: str) -> None: tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: f.write(text) os.replace(tmp, path) def main() -> int: if not os.path.exists(JSON_PATH): log({"event": "skip", "reason": "no_state_file"}) return 0 with open(JSON_PATH, encoding="utf-8") as f: state = json.load(f) orders = state.get("orders", []) if not orders: log({"event": "skip", "reason": "no_orders"}) return 0 # 환율 (USD->KRW) 실시간 스팟 + 증권사 기준환율 보정(FX_SPREAD) try: spot = yahoo_price("KRW=X") except Exception as e: spot = None log({"event": "fx_fail", "err": str(e)[:120]}) spread = read_fx_spread() override = read_fx_override() if override: fx = round(float(override), 4); fx_source = "기준환율(고정)" elif spot: fx = round(float(spot) * spread, 4); fx_source = "시장스팟" + ("×보정" if spread != 1.0 else "") else: fx = None; fx_source = "없음" updated, failed = [], [] prices: dict = {} for o in orders: sym = o.get("ticker") qty = o.get("qty") if not sym or not qty: continue try: price = yahoo_price(sym) if not price: raise ValueError("no price") except Exception as e: failed.append(sym) log({"event": "price_fail", "ticker": sym, "err": str(e)[:120]}) continue old_value = o.get("value_krw") old_pnl = o.get("pnl_krw") # 원가(KRW) = 기존 평가액 - 기존 평가손익 (수량·평단 불변이므로 상수) buy_basis = (old_value - old_pnl) if isinstance(old_value, (int, float)) and isinstance(old_pnl, (int, float)) else None o["current_price"] = round(float(price), 4) prices[sym] = float(price) # ── 목표까지 거리·예상손익·상태 라이브 재계산 (현재가·환율 기준) ── c = float(price) def _gp(t): # 현재가→목표 수익률% return round((float(t) - c) / c * 100, 1) if t not in (None, "") else None def _ep(t): # 예상 손익(KRW) return round(qty * (float(t) - c) * float(fx)) if (t not in (None, "") and fx) else None if o.get("t1") is not None: o["t1_gain_pct"] = _gp(o["t1"]); o["t1_est_profit_krw"] = _ep(o["t1"]) if o.get("t2") is not None: o["t2_gain_pct"] = _gp(o["t2"]); o["t2_est_profit_krw"] = _ep(o["t2"]) if o.get("hold_until") is not None: o["hold_until_gain_pct"] = _gp(o["hold_until"]); o["hold_until_est_profit_krw"] = _ep(o["hold_until"]) if o.get("prior_high") is not None: o["prior_high_gap_pct"] = _gp(o["prior_high"]) if o.get("invalid") is not None: o["invalid_loss_pct"] = _gp(o["invalid"]); o["invalid_est_loss_krw"] = _ep(o["invalid"]) t1, t2, inv = o.get("t1"), o.get("t2"), o.get("invalid") st = "진행" if inv and c <= float(inv): st = "손절이탈" elif t2 and c >= float(t2): st = "T2도달" elif t1 and c >= float(t1): st = "T1돌파" elif t1 and (float(t1) - c) / c <= 0.015: st = "T1근접" elif inv and (c - float(inv)) / c <= 0.015: st = "손절근접" o["target_state"] = st # 환율 결정: yahoo 환율 우선, 없으면 기존값에서 역산 use_fx = fx if use_fx is None and isinstance(old_value, (int, float)) and o.get("_prev_price"): try: use_fx = old_value / (qty * o["_prev_price"]) except Exception: use_fx = None if use_fx: new_value = round(qty * float(price) * float(use_fx)) o["value_krw"] = new_value if buy_basis is not None: o["pnl_krw"] = new_value - buy_basis updated.append(sym) now = dt.datetime.now(dt.timezone.utc).astimezone() state["prices_updated_at"] = now.isoformat() if spot: state["fx_market"] = round(float(spot), 2) # 시장 실시간 스팟(참고) if fx: state["fx_usdkrw"] = round(float(fx), 2) # 평가에 실제 적용된 환율 state["fx_source"] = fx_source # 적용 환율 출처 # 계좌별(본/부) 산출 라이브 갱신 acct_res = update_accounts(state, prices, fx) log({"event": "accounts_update", **acct_res}) text = json.dumps(state, ensure_ascii=False, indent=2) atomic_write(JSON_PATH, text) atomic_write(JS_PATH, JS_PREFIX + text + ";\n") log({"event": "done", "updated": updated, "failed": failed, "fx": fx}) print(f"가격 갱신 완료: {len(updated)}종목, 실패 {failed or '없음'}, 환율 {fx}") return 0 if __name__ == "__main__": raise SystemExit(main())