Spaces:
Running
Running
| """ | |
| stock.py — "Did this stock actually change?" on free Yahoo data (no API key). | |
| Finance-native use of the same machinery, honestly matched to how prices behave: | |
| * SHOCK events — the Clutch's MagnitudeGate fed with |daily return| / typical move. | |
| A single outsized day trips it instantly; several stressed days in a | |
| row accumulate and trip it too (which a naive threshold misses). | |
| * VOLATILITY regime changes — the iid change detector (change.detect) run on rolling | |
| daily volatility: "a normal day used to be ±1.1%, now it's ±2.6%". | |
| * DRIFT — total return over the window, compared against what pure luck could | |
| produce (sigma * sqrt(n)), stated plainly. | |
| Data: yfinance (Yahoo Finance scrape — free, keyless). News: yfinance's news feed, | |
| also keyless. Both can rate-limit; failures are reported, never faked. | |
| """ | |
| import time | |
| import numpy as np | |
| from clutch import Clutch, MagnitudeGate | |
| from change import detect | |
| # ------------------------------------------------------------------ data (keyless) | |
| _CACHE = {} | |
| _TTL = 900 # 15 min | |
| def fetch_prices(ticker, period="1y"): | |
| """Returns (dates, close, currency, err). Cached to be polite to Yahoo.""" | |
| key = (ticker.upper().strip(), period, int(time.time() // _TTL)) | |
| if key in _CACHE: | |
| return _CACHE[key] | |
| try: | |
| import yfinance as yf | |
| tk = yf.Ticker(ticker.strip()) | |
| hist = tk.history(period=period, interval="1d", auto_adjust=True) | |
| if hist is None or len(hist) < 40 or "Close" not in hist: | |
| out = (None, None, "", f"Couldn't get enough daily data for '{ticker}'. " | |
| "Check the symbol (Yahoo format, e.g. AAPL, NOK, BTC-USD, ^GSPC).") | |
| else: | |
| close = hist["Close"].to_numpy(dtype=float) | |
| dates = [d.strftime("%Y-%m-%d") for d in hist.index] | |
| cur = "" | |
| try: | |
| cur = tk.fast_info.get("currency") or "" | |
| except Exception: | |
| pass | |
| out = (dates, close, cur, None) | |
| except Exception as e: | |
| out = (None, None, "", f"Data fetch failed ({type(e).__name__}). Yahoo sometimes " | |
| "rate-limits shared servers — wait a minute and try again.") | |
| _CACHE[key] = out | |
| return out | |
| def fetch_news(ticker, k=6): | |
| """Free Yahoo headlines via yfinance; tolerant of old and new item formats.""" | |
| try: | |
| import yfinance as yf | |
| raw = yf.Ticker(ticker.strip()).news or [] | |
| except Exception: | |
| return [] | |
| return parse_news(raw, k) | |
| def parse_news(raw, k=6): | |
| items = [] | |
| for it in raw[: k * 2]: | |
| c = it.get("content", it) if isinstance(it, dict) else {} | |
| title = c.get("title") or it.get("title") | |
| if not title: | |
| continue | |
| url = "" | |
| cu = c.get("canonicalUrl") or c.get("clickThroughUrl") or {} | |
| if isinstance(cu, dict): | |
| url = cu.get("url", "") | |
| url = url or it.get("link", "") | |
| prov = c.get("provider") or {} | |
| publisher = (prov.get("displayName") if isinstance(prov, dict) else None) \ | |
| or it.get("publisher", "") | |
| when = c.get("pubDate") or c.get("displayTime") or "" | |
| if not when and it.get("providerPublishTime"): | |
| when = time.strftime("%Y-%m-%d", time.gmtime(it["providerPublishTime"])) | |
| when = str(when)[:10] | |
| items.append(dict(title=title.strip(), url=url, publisher=publisher, when=when)) | |
| if len(items) >= k: | |
| break | |
| return items | |
| # ------------------------------------------------------------------ analysis | |
| def robust_sigma(x): | |
| return float(1.4826 * np.median(np.abs(x - np.median(x))) + 1e-12) | |
| def local_sigma(rets, win=60, warm=20): | |
| """Past-only rolling robust sigma, so a rough regime stops spamming shock flags | |
| but a fresh crash (judged against the calm past) still screams.""" | |
| g = robust_sigma(rets) | |
| out = np.full(len(rets), g) | |
| for i in range(warm, len(rets)): | |
| out[i] = max(robust_sigma(rets[max(0, i - win):i]), 0.4 * g) | |
| return out | |
| def shock_events(rets, sensitivity=1.0): | |
| """Clutch gate on |return| / local typical move. A single outsized day trips it | |
| instantly; several stressed days in a row accumulate and trip it too. | |
| Returns (events, sig_global, calm_days).""" | |
| sig = robust_sigma(rets) | |
| loc = local_sigma(rets) | |
| gate = MagnitudeGate(gain=3.0, leak=3.2, trip=8.0 * sensitivity) | |
| trips = [] | |
| for i, r in enumerate(rets): | |
| if gate.update(abs(r) / loc[i]): | |
| trips.append(i) | |
| gate.clear() | |
| events = [] | |
| for i in trips: | |
| if events and i - events[-1][-1] <= 3: | |
| events[-1].append(i) | |
| else: | |
| events.append([i]) | |
| out = [] | |
| for grp in events: | |
| i0 = max(0, grp[0] - 2) | |
| i1 = min(len(rets) - 1, grp[-1]) | |
| cum = float(np.sum(rets[i0:i1 + 1])) | |
| peak_i = i0 + int(np.argmax(np.abs(rets[i0:i1 + 1]))) | |
| peak = float(np.abs(rets[peak_i]) / loc[peak_i]) | |
| biggest = float(rets[peak_i]) | |
| out.append(dict(i0=i0, i1=i1, cum_ret=cum, peak_z=peak, biggest=biggest)) | |
| calm = len(rets) - sum(e["i1"] - e["i0"] + 1 for e in out) | |
| return out, sig, calm | |
| def vol_regimes(rets, sensitivity=1.0, win=15): | |
| """Non-overlapping window vols + strongest-split ratio test (near-independent | |
| samples, unlike a rolling window). Reports at most one regime change — the | |
| strongest — per period. Returns [] or [dict(at, before, after, ratio)].""" | |
| n = len(rets) | |
| m = n // win | |
| if m < 8: | |
| return [] | |
| v = np.array([robust_sigma(rets[k * win:(k + 1) * win]) for k in range(m)]) | |
| best = None | |
| for k in range(3, m - 2): | |
| before, after = float(np.median(v[:k])), float(np.median(v[k:])) | |
| ratio = after / (before + 1e-12) | |
| score = max(ratio, 1.0 / ratio) | |
| if best is None or score > best[0]: | |
| best = (score, k, before, after, ratio) | |
| score, k, before, after, ratio = best | |
| thresh = 1.0 + 0.75 * sensitivity # sens 1 -> 1.75x; strict 1.5 -> ~2.1x; eager 0.7 -> ~1.5x | |
| if score < thresh: | |
| return [] | |
| # refine the change day: best day-level split within +/-2 windows of the coarse one | |
| c0 = k * win | |
| best_c, best_dev = c0, 0.0 | |
| for c in range(max(30, c0 - 2 * win), min(n - 30, c0 + 2 * win)): | |
| b = robust_sigma(rets[max(0, c - 60):c]) | |
| a = robust_sigma(rets[c:c + 60]) | |
| dev = abs(np.log(a / (b + 1e-12) + 1e-12)) | |
| if dev > best_dev: | |
| best_dev, best_c = dev, c | |
| before = robust_sigma(rets[max(0, best_c - 60):best_c]) | |
| after = robust_sigma(rets[best_c:best_c + 60]) | |
| return [dict(at=best_c, before=before, after=after, ratio=after / (before + 1e-12))] | |
| def analyze(dates, close, sensitivity=1.0): | |
| logp = np.log(np.asarray(close, float)) | |
| rets = np.diff(logp) | |
| shocks, sig_d, calm = shock_events(rets, sensitivity) | |
| vols = vol_regimes(rets, sensitivity) | |
| total = float(logp[-1] - logp[0]) | |
| luck2 = 2.0 * sig_d * np.sqrt(len(rets)) # 2-sigma of pure-luck drift | |
| return dict(rets=rets, sig_d=sig_d, shocks=shocks, vols=vols, calm=calm, | |
| total=total, luck2=luck2, n=len(rets)) | |
| # ------------------------------------------------------------------ language | |
| def pct(x): | |
| return f"{(np.exp(x) - 1) * 100:+.1f}%" | |
| def verdict_md(ticker, dates, close, a, currency=""): | |
| n = a["n"] | |
| d0, d1 = dates[0], dates[-1] | |
| sig_pct = (np.exp(a["sig_d"]) - 1) * 100 | |
| lines = [] | |
| if not a["shocks"] and not a["vols"]: | |
| lines.append(f"## 😌 {ticker}: a quiet stretch — ordinary wobble only") | |
| lines.append(f"Across **{n} trading days** ({d0} → {d1}), no day or cluster of days " | |
| f"broke out of this stock's normal movement (a typical day here is about " | |
| f"**±{sig_pct:.1f}%**). Every scary-looking dip in this window was " | |
| f"within what dice would produce.") | |
| else: | |
| k = len(a["shocks"]) + len(a["vols"]) | |
| lines.append(f"## 🔔 {ticker}: {k} real event{'s' if k != 1 else ''} in this window") | |
| ranked = sorted(a["shocks"], | |
| key=lambda e: max(e["peak_z"], abs(e["cum_ret"]) / (a["sig_d"] + 1e-12)), | |
| reverse=True) | |
| hidden = max(0, len(ranked) - 5) | |
| for e in sorted(ranked[:5], key=lambda e: e["i0"]): | |
| day0, day1 = dates[e["i0"] + 1], dates[e["i1"] + 1] | |
| span = f"on {day1}" if e["i0"] == e["i1"] else f"over {day0} → {day1}" | |
| if abs(e["cum_ret"]) < 0.5 * abs(e["biggest"]): | |
| lines.append(f"- **Violent swings {span}** that largely cancelled out " | |
| f"(net {pct(e['cum_ret'])}, sharpest single day {pct(e['biggest'])}, " | |
| f"about {e['peak_z']:.0f}× normal). Something happened there even " | |
| f"though the price ended near where it started.") | |
| else: | |
| direction = "down" if e["cum_ret"] < 0 else "up" | |
| lines.append(f"- **Shock {span}**: moved **{direction} {pct(e['cum_ret'])}** " | |
| f"(sharpest day about {e['peak_z']:.0f}× a normal day). That is a " | |
| f"real event, not wobble — worth knowing *why* (headlines below).") | |
| if hidden: | |
| lines.append(f"- (+ {hidden} smaller flare-up{'s' if hidden > 1 else ''} not " | |
| f"listed — nothing above {ranked[5]['peak_z']:.0f}× normal.)") | |
| for v in a["vols"]: | |
| day = dates[min(v["at"] + 1, len(dates) - 1)] | |
| b = (np.exp(v["before"]) - 1) * 100 | |
| af = (np.exp(v["after"]) - 1) * 100 | |
| word = "rougher" if af > b else "calmer" | |
| lines.append(f"- **The ride got {word} around {day}**: a typical day went from " | |
| f"about ±{b:.1f}% to ±{af:.1f}%. Same stock, different weather.") | |
| # drift vs luck — the part people get wrong most | |
| tot, luck = a["total"], a["luck2"] | |
| lines.append("") | |
| if abs(tot) > luck: | |
| lines.append(f"**The drift is real too:** {pct(tot)} over the period — more than the " | |
| f"±{(np.exp(luck)-1)*100:.0f}% that pure day-to-day luck could plausibly " | |
| f"produce over {n} days.") | |
| else: | |
| lines.append(f"**About the overall {pct(tot)} move:** over {n} days, pure luck at this " | |
| f"stock's wobble could produce anything within about " | |
| f"±{(np.exp(luck)-1)*100:.0f}%. So the period's drift, by itself, is " | |
| f"**not distinguishable from chance** — an honest thing almost no chart " | |
| f"commentary will tell you.") | |
| lines.append("") | |
| lines.append(f"**Your attention, saved:** {a['calm']} of {n} days were inside the normal " | |
| f"band — days when checking the chart could tell you nothing.") | |
| lines.append("") | |
| lines.append("<small>This describes what already happened in free Yahoo data; it predicts " | |
| "nothing and is not investment advice or a recommendation to buy or sell " | |
| "anything. Past shocks say nothing about future ones.</small>") | |
| return "\n".join(lines) | |