#!/usr/bin/env python3 """Clean recompute for 6669 + 00412A + Asia-semi context, dropping phantom zero-volume bars and checking the 6669 9/2 ex-rights discontinuity.""" import json, ssl, time, urllib.request, datetime as dt import pandas as pd, numpy as np ctx = ssl.create_default_context(); ctx.check_hostname = False; ctx.verify_mode = ssl.CERT_NONE UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124.0 Safari/537.36"} OUT = {} def yq(ticker, rng="6mo"): for host in ("query1", "query2"): try: u = (f"https://{host}.finance.yahoo.com/v8/finance/chart/{ticker}" f"?range={rng}&interval=1d&includePrePost=false&events=div%2Csplit") d = json.loads(urllib.request.urlopen(urllib.request.Request(u, headers=UA), timeout=30, context=ctx).read().decode()) r = d["chart"]["result"][0]; q = r["indicators"]["quote"][0] idx = pd.to_datetime(r["timestamp"], unit="s").tz_localize("UTC").tz_convert("Asia/Taipei").normalize() df = pd.DataFrame(q, index=idx) for c in df.columns: df[c] = pd.to_numeric(df[c], errors="coerce") adj = r["indicators"].get("adjclose") df["adjclose"] = pd.to_numeric(adj[0]["adjclose"], errors="coerce") if adj else np.nan df = df[~df.index.duplicated(keep="last")].sort_index() return df except Exception as e: print(f" !! {ticker}@{host}: {type(e).__name__}: {e}", flush=True) time.sleep(2) return None def tech(df): d = df.dropna(subset=["close"]).copy() d = d[d["volume"].fillna(0) > 0] # drop phantom zero-volume bars c, h, l, v = d["close"], d["high"], d["low"], d["volume"] d["ma20"] = c.rolling(20).mean(); d["ma60"] = c.rolling(60).mean() delta = c.diff() gain = delta.clip(lower=0).ewm(alpha=1/14, adjust=False, min_periods=14).mean() loss = (-delta.clip(upper=0)).ewm(alpha=1/14, adjust=False, min_periods=14).mean() rs = gain / loss d["rsi"] = (100 - 100/(1+rs)).replace([np.inf,-np.inf], np.nan).fillna(50) e12 = c.ewm(span=12, adjust=False).mean(); e26 = c.ewm(span=26, adjust=False).mean() d["macd"] = e12-e26; d["signal"] = d["macd"].ewm(span=9, adjust=False).mean(); d["hist"] = d["macd"]-d["signal"] ll = l.rolling(9).min(); hh = h.rolling(9).max() rsv = (c-ll)/(hh-ll)*100 d["k"] = rsv.ewm(alpha=1/3, adjust=False).mean(); d["d"] = d["k"].ewm(alpha=1/3, adjust=False).mean() d["vma20"] = v.rolling(20).mean() d["low20"] = l.rolling(20).min(); d["high60"] = h.rolling(60).max() return d # ---------- 6669 ---------- print("== 6669 ==", flush=True) raw = yq("6669.TW", "6mo") if raw is not None: print(" raw bars:", len(raw), "last:", raw.index[-1].date(), flush=True) # ex-rights discontinuity check on raw close s = raw["close"].dropna() jump = s.pct_change().abs() > 0.25 for i, v in s[jump].items(): print(f" !! raw close jump {i.date()}: {s.shift(1)[i]:.1f} -> {v:.1f} ({v/s.shift(1)[i]-1:+.1%})", flush=True) print(f" raw max close: {s.max():.1f} on {s.idxmax().date()}", flush=True) if "adjclose" in raw and raw["adjclose"].notna().any(): a = raw["adjclose"].dropna() print(f" adjclose max: {a.max():.1f} on {a.idxmax().date()}; last adj {a.iloc[-1]:.2f} vs close {raw['close'].dropna().iloc[-1]:.2f}", flush=True) d = tech(raw) last, prev = d.iloc[-1], d.iloc[-2] rows = [] for i, r in d.tail(12).iterrows(): rows.append({"d": str(i.date()), "c": float(r["close"]), "v": int(r["volume"]), "ma20": round(float(r["ma20"]),1), "ma60": round(float(r["ma60"]),1), "rsi": round(float(r["rsi"]),1), "hist": round(float(r["hist"]),2), "k": round(float(r["k"]),1), "d": round(float(r["d"]),1), "vma20": int(r["vma20"]), "low20": round(float(r["low20"]),1), "below60": bool(r["close"] < r["ma60"])}) OUT["6669"] = {"last_date": str(d.index[-1].date()), "bars": int(len(d)), "close": float(last["close"]), "prev_close": float(prev["close"]), "chg_pct": float((last["close"]/prev["close"]-1)*100), "ma20": float(last["ma20"]), "ma60": float(last["ma60"]), "rsi": round(float(last["rsi"]),2), "macd": float(last["macd"]), "signal": float(last["signal"]), "hist": float(last["hist"]), "k": round(float(last["k"]),2), "d": round(float(last["d"]),2), "prev_k": round(float(prev["k"]),2), "prev_d": round(float(prev["d"]),2), "volume": int(last["volume"]), "vma20": float(last["vma20"]), "vol_ratio": float(last["volume"]/last["vma20"]), "low20": float(last["low20"]), "high60": float(last["high60"]), "from_high60_pct": float((last["close"]/last["high60"]-1)*100), "from_ma20_pct": float((last["close"]/last["ma20"]-1)*100), "from_ma60_pct": float((last["close"]/last["ma60"]-1)*100), "tail": rows} print(" ", OUT["6669"]["last_date"], OUT["6669"]["close"], "RSI", OUT["6669"]["rsi"], "MA20", round(OUT["6669"]["ma20"],1), "MA60", round(OUT["6669"]["ma60"],1), "volratio", round(OUT["6669"]["vol_ratio"],2), flush=True) n_below = 0 for r in reversed(rows): if r["below60"]: n_below += 1 else: break OUT["6669"]["consec_below_ma60"] = n_below print(" consecutive closes below MA60:", n_below, flush=True) # ---------- 00412A ---------- print("== 00412A ==", flush=True) a_raw = yq("00412A.TW", "6mo") if a_raw is not None: a = a_raw.dropna(subset=["close"]).copy() a = a[a["volume"].fillna(0) > 0] a = a[a["close"] > 0] al = a.iloc[-1] OUT["00412A"] = {"last_date": str(a.index[-1].date()), "bars": int(len(a)), "first_date": str(a.index[0].date()), "close": float(al["close"]), "volume": int(al["volume"]), "prev_close": float(a["close"].iloc[-2]), "prev_date": str(a.index[-2].date()), "chg_pct": float((al["close"]/a["close"].iloc[-2]-1)*100), "since_first_pct": float((al["close"]/a["close"].iloc[0]-1)*100), "first_close": float(a["close"].iloc[0]), "high": float(a["close"].max()), "low": float(a["close"].min()), "hist": [{"d": str(i.date()), "c": float(r["close"]), "v": int(r["volume"])} for i, r in a.iterrows()]} print(" ", OUT["00412A"]["last_date"], OUT["00412A"]["close"], "bars", OUT["00412A"]["bars"], "since first", round(OUT["00412A"]["since_first_pct"],1), "%", flush=True) # ---------- Asia-semi context ---------- print("== Asia context ==", flush=True) OUT["asia"] = {} for t, label in [("^SOX","SOX"), ("0981.HK","SMIC"), ("005930.KS","Samsung"), ("TSM","TSM ADR"), ("JPYTWD=X","JPY/TWD"), ("^N225","Nikkei"), ("^HSI","Hang Seng"), ("2330.TW","TSMC TW"), ("2344.TW","華邦電"), ("2379.TW","瑞昱"), ("2308.TW","台達電"), ("2882.TW","國泰金"), ("NVDA","NVDA"), ("CRWV","CoreWeave")]: x = yq(t, "3mo") if x is None or len(x) < 2: OUT["asia"][label] = None; continue x = x.dropna(subset=["close"]) if len(x) < 2: OUT["asia"][label] = None; continue l = x["close"].iloc[-1] e = {"last_date": str(x.index[-1].date()), "close": float(l), "chg_pct": float((l/x["close"].iloc[-2]-1)*100), "chg5_pct": float((l/x["close"].iloc[-6]-1)*100) if len(x) >= 6 else None, "chg3_pct": float((l/x["close"].iloc[-4]-1)*100) if len(x) >= 4 else None} if t == "^SOX": h60 = x["high"].rolling(60).max().iloc[-1] e["high60"] = float(h60); e["from_high60_pct"] = float((l/h60-1)*100) OUT["asia"][label] = e print(" ", label, e["last_date"], round(e["close"],3), f'{e["chg_pct"]:+.2f}%', flush=True) json.dump(OUT, open("/root/.hermes/cron/output/_clean_20261026.json","w"), ensure_ascii=False, indent=1) print("SAVED")