Spaces:

s-ttp
/

crucible

Paused

AnalysisState pipeline: reconciliation, staged validation, frozen state, claim-token prose, audited renderer; public/QA split

c9fa753 verified about 2 months ago

Raw History Blame Contribute Delete

10.3 kB

"""Crucible — AI Equity Analyst Space.
A ticker in → a live 4-role LLM debate (Neutral · Bull · Bear · Portfolio Manager) over a
quant ML forecast → a Buy/Hold/Sell verdict. The heavy engine runs as a subprocess and
streams to the browser over SSE.
"""
import json
import os
import threading
from datetime import datetime, timezone

from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import FileResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles

import config
import analyst

app = FastAPI(title="Crucible")

# ── Abuse guard: global per-UTC-day run cap ─────────────────────────────────
# The Space is public, so anyone can trigger runs — and every run spends real
# LLM + Alpha Vantage budget. Cap the number of analyses started per day (UTC).
# Tune via the CRUCIBLE_DAILY_CAP env/secret; a run reads/writes a small counter
# file so the limit survives within the container's lifetime.
DAILY_CAP = int(os.getenv("CRUCIBLE_DAILY_CAP", "30"))
_CAP_FILE = os.getenv("CRUCIBLE_CAP_FILE", "/tmp/crucible_runcount.json")
_cap_lock = threading.Lock()


def _reserve_run() -> tuple[bool, int, int]:
    """Atomically count one run against today's cap. Returns (allowed, used, cap)."""
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    with _cap_lock:
        data = {"date": today, "count": 0}
        try:
            with open(_CAP_FILE) as f:
                prev = json.load(f)
            if prev.get("date") == today:
                data = prev
        except Exception:
            pass
        if data["count"] >= DAILY_CAP:
            return False, data["count"], DAILY_CAP
        data["count"] += 1
        try:
            with open(_CAP_FILE, "w") as f:
                json.dump(data, f)
        except Exception:
            pass
        return True, data["count"], DAILY_CAP


@app.get("/api/health")
def health():
    import os
    return {"app": "Crucible",
            "engine": (config.RECOMMENDER_DIR / "predict_yearly.py").exists(),
            "llm": {k: bool(os.getenv(k)) for k in
                    ("GEMINI_API_KEY", "OPENAI_API_KEY", "KIMI_API_KEY", "ANTHROPIC_API_KEY")},
            "data": bool(os.getenv("ALPHAVANTAGE_API_KEY") or os.getenv("ALPHA_VANTAGE_API_KEY")),
            "panel_ml": (config.RECOMMENDER_DIR / "panel_ml_snapshot.parquet").exists()}


@app.post("/api/analyst/{ticker}")
def analyst_start(ticker: str, request: Request, x_crucible_client: str = Header(default=""),
                  x_prewarm_token: str = Header(default="")):
    # Only the Crucible web page (or the token-holding prewarm job) may start a run. The
    # Space is public and every run spends LLM budget; an anonymous script posted an AAPL
    # analysis on 2026-09-06 with no page load and never read the result. The header is
    # trivial to copy, but it stops crawlers and API guessers. Who started a run is recorded
    # ONLY in the private server-side access log (analyst.log_access) — never in the report,
    # the API response, the saved report, the markdown or the report dataset.
    _tok = os.getenv("PREWARM_TOKEN", "")
    if x_crucible_client != "ui" and not (_tok and x_prewarm_token == _tok):
        raise HTTPException(status_code=403, detail="analyses can only be started from the Crucible web page")
    _via = "ui" if x_crucible_client == "ui" else "token"
    # Resolve FIRST (ticker or company name -> verified symbol) so a typo neither burns a
    # daily-cap slot nor launches an engine run that dies with a raw "failed (exit 1)".
    try:
        res = analyst.resolve_symbol(ticker)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    analyst._log_request(res["symbol"])
    allowed, _used, cap = _reserve_run()
    if not allowed:
        raise HTTPException(
            status_code=429,
            detail=f"Daily analysis limit reached ({cap}/day). This resets at 00:00 UTC — please try again then.")
    try:
        out = {"job_id": analyst.start(res["symbol"]), "ticker": res["symbol"]}
        _j = analyst._jobs.get(out["job_id"])
        if _j is not None and not _j.get("source"):
            _j["source"] = _via
        analyst.log_access(out["job_id"], res["symbol"], _via, request)
        if res.get("resolved_from"):
            out["resolved"] = {"from": res["resolved_from"], "name": res.get("name")}
        return out
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))


@app.get("/api/track-record")
def track_record(min_age_days: int = 30):
    """Realized accuracy of past calls (direction hit-rate, mean abs error) once they age."""
    return analyst.track_record(min_age_days=min_age_days)


@app.get("/api/recent")
def recent():
    """Cached analyses from the last 7 days — powers the Recent chips on the landing page."""
    return {"recent": analyst.recent_reports()}


@app.get("/api/report/{ticker}/md")
def cached_report_md(ticker: str):
    """The stored audited markdown: the same fragments as the web page, byte for byte."""
    from fastapi.responses import PlainTextResponse
    md = analyst.cached_markdown(ticker)
    if not md:
        raise HTTPException(status_code=404, detail="no published analysis for this ticker")
    return PlainTextResponse(md, media_type="text/markdown; charset=utf-8",
                             headers={"Content-Disposition": f'attachment; filename="Crucible_{ticker.upper()}.md"'})


@app.get("/api/report/{ticker}")
def cached_report(ticker: str):
    """Serve a cached full analysis instantly (0 LLM calls, no engine lock). 404 when absent
    or older than 7 days; includes a staleness flag when the price has drifted >7% since."""
    d = analyst.cached_report(ticker)
    if not d:
        raise HTTPException(status_code=404, detail="no published analysis for this ticker")
    return d


@app.get("/api/dev/qa/{ticker}")
def dev_qa(ticker: str, x_dev_token: str = Header(default="")):
    """Internal QA record (validation results, rejected sentences, state snapshot). Developer-only:
    requires the DEV_TOKEN secret; without it configured the route does not exist."""
    tok = os.getenv("DEV_TOKEN", "")
    if not tok:
        raise HTTPException(status_code=404, detail="not found")
    import hmac
    if not hmac.compare_digest(x_dev_token, tok):
        raise HTTPException(status_code=403, detail="forbidden")
    rec = analyst.qa_record(ticker)
    if rec is None:
        raise HTTPException(status_code=404, detail="no QA record for this ticker")
    return rec


@app.post("/api/prewarm")
def prewarm(x_prewarm_token: str = Header(default="")):
    """Weekly, gated cache-warmer: only fires when organic searches haven't already filled
    the Recent row (>=3 fresh entries -> no-op). Token-gated so public callers can't spend
    LLM budget. Picks news-active large-caps not already cached, max 3."""
    import os as _o
    tok = _o.getenv("PREWARM_TOKEN", "")
    if not tok or x_prewarm_token != tok:
        raise HTTPException(status_code=403, detail="bad or missing prewarm token")
    fresh = analyst.recent_reports(limit=50)
    if len(fresh) >= 3:
        return {"started": [], "reason": f"organic cache sufficient ({len(fresh)} fresh entries)"}
    need = 3 - len(fresh)
    have = {r["ticker"] for r in fresh}
    started = []
    try:
        import urllib.request as _u, json as _j
        av = _o.getenv("ALPHAVANTAGE_API_KEY") or _o.getenv("ALPHA_VANTAGE_API_KEY", "")
        # What's trending IN THE NEWS: aggregate relevance-weighted ticker mentions across
        # the latest news feed (not "most traded", which skews to meme/penny volume).
        with _u.urlopen("https://www.alphavantage.co/query?function=NEWS_SENTIMENT"
                        f"&sort=LATEST&limit=200&apikey={av}", timeout=25) as r:
            feed = _j.loads(r.read()).get("feed") or []
        score: dict = {}
        for a in feed:
            for ts in a.get("ticker_sentiment") or []:
                sym = (ts.get("ticker") or "").upper()
                if ":" in sym or not sym.isalpha() or len(sym) > 5:   # skip CRYPTO:/FOREX:/funds
                    continue
                score[sym] = score.get(sym, 0.0) + float(ts.get("relevance_score") or 0)
        for sym, _sc in sorted(score.items(), key=lambda x: -x[1]):
            if len(started) >= need:
                break
            if sym in have:
                continue
            try:
                res = analyst.resolve_symbol(sym)     # verifies a real quote exists
                jid = analyst.start(res["symbol"])
                analyst._jobs[jid]["source"] = "prewarmed"
                started.append(res["symbol"]); have.add(res["symbol"])
            except ValueError:
                continue
    except Exception as exc:
        return {"started": started, "error": str(exc)[:120]}
    return {"started": started, "basis": "news-trending (relevance-weighted mentions, latest feed)"}


@app.get("/api/jobs")
def jobs():
    """Who holds the engine, who is queued, recent jobs — for diagnosing 'stuck' runs."""
    return analyst.jobs_view()


@app.get("/api/usage")
def usage():
    """Today's run count vs the daily cap (no side effects)."""
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    used = 0
    try:
        with open(_CAP_FILE) as f:
            d = json.load(f)
        if d.get("date") == today:
            used = d.get("count", 0)
    except Exception:
        pass
    return {"date": today, "used": used, "cap": DAILY_CAP}


@app.get("/api/analyst/{job_id}/stream")
def analyst_stream(job_id: str):
    if not analyst.exists(job_id):
        raise HTTPException(status_code=404, detail="unknown job")
    return StreamingResponse(analyst.stream(job_id), media_type="text/event-stream")


@app.get("/api/analyst/{job_id}/result")
def analyst_result(job_id: str):
    if not analyst.exists(job_id):
        raise HTTPException(status_code=404, detail="unknown job")
    return analyst.result(job_id)


@app.get("/")
def index():
    return FileResponse(config.APP_DIR / "static" / "index.html", headers={"Cache-Control": "no-cache, must-revalidate"})


app.mount("/static", StaticFiles(directory=config.APP_DIR / "static"), name="static")

	"""Crucible — AI Equity Analyst Space.

	A ticker in → a live 4-role LLM debate (Neutral · Bull · Bear · Portfolio Manager) over a
	quant ML forecast → a Buy/Hold/Sell verdict. The heavy engine runs as a subprocess and
	streams to the browser over SSE.
	"""
	import json
	import os
	import threading
	from datetime import datetime, timezone

	from fastapi import FastAPI, Header, HTTPException, Request
	from fastapi.responses import FileResponse, StreamingResponse
	from fastapi.staticfiles import StaticFiles

	import config
	import analyst

	app = FastAPI(title="Crucible")

	# ── Abuse guard: global per-UTC-day run cap ─────────────────────────────────
	# The Space is public, so anyone can trigger runs — and every run spends real
	# LLM + Alpha Vantage budget. Cap the number of analyses started per day (UTC).
	# Tune via the CRUCIBLE_DAILY_CAP env/secret; a run reads/writes a small counter
	# file so the limit survives within the container's lifetime.
	DAILY_CAP = int(os.getenv("CRUCIBLE_DAILY_CAP", "30"))
	_CAP_FILE = os.getenv("CRUCIBLE_CAP_FILE", "/tmp/crucible_runcount.json")
	_cap_lock = threading.Lock()


	def _reserve_run() -> tuple[bool, int, int]:
	"""Atomically count one run against today's cap. Returns (allowed, used, cap)."""
	today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
	with _cap_lock:
	data = {"date": today, "count": 0}
	try:
	with open(_CAP_FILE) as f:
	prev = json.load(f)
	if prev.get("date") == today:
	data = prev
	except Exception:
	pass
	if data["count"] >= DAILY_CAP:
	return False, data["count"], DAILY_CAP
	data["count"] += 1
	try:
	with open(_CAP_FILE, "w") as f:
	json.dump(data, f)
	except Exception:
	pass
	return True, data["count"], DAILY_CAP


	@app.get("/api/health")
	def health():
	import os
	return {"app": "Crucible",
	"engine": (config.RECOMMENDER_DIR / "predict_yearly.py").exists(),
	"llm": {k: bool(os.getenv(k)) for k in
	("GEMINI_API_KEY", "OPENAI_API_KEY", "KIMI_API_KEY", "ANTHROPIC_API_KEY")},
	"data": bool(os.getenv("ALPHAVANTAGE_API_KEY") or os.getenv("ALPHA_VANTAGE_API_KEY")),
	"panel_ml": (config.RECOMMENDER_DIR / "panel_ml_snapshot.parquet").exists()}


	@app.post("/api/analyst/{ticker}")
	def analyst_start(ticker: str, request: Request, x_crucible_client: str = Header(default=""),
	x_prewarm_token: str = Header(default="")):
	# Only the Crucible web page (or the token-holding prewarm job) may start a run. The
	# Space is public and every run spends LLM budget; an anonymous script posted an AAPL
	# analysis on 2026-09-06 with no page load and never read the result. The header is
	# trivial to copy, but it stops crawlers and API guessers. Who started a run is recorded
	# ONLY in the private server-side access log (analyst.log_access) — never in the report,
	# the API response, the saved report, the markdown or the report dataset.
	_tok = os.getenv("PREWARM_TOKEN", "")
	if x_crucible_client != "ui" and not (_tok and x_prewarm_token == _tok):
	raise HTTPException(status_code=403, detail="analyses can only be started from the Crucible web page")
	_via = "ui" if x_crucible_client == "ui" else "token"
	# Resolve FIRST (ticker or company name -> verified symbol) so a typo neither burns a
	# daily-cap slot nor launches an engine run that dies with a raw "failed (exit 1)".
	try:
	res = analyst.resolve_symbol(ticker)
	except ValueError as exc:
	raise HTTPException(status_code=400, detail=str(exc))
	analyst._log_request(res["symbol"])
	allowed, _used, cap = _reserve_run()
	if not allowed:
	raise HTTPException(
	status_code=429,
	detail=f"Daily analysis limit reached ({cap}/day). This resets at 00:00 UTC — please try again then.")
	try:
	out = {"job_id": analyst.start(res["symbol"]), "ticker": res["symbol"]}
	_j = analyst._jobs.get(out["job_id"])
	if _j is not None and not _j.get("source"):
	_j["source"] = _via
	analyst.log_access(out["job_id"], res["symbol"], _via, request)
	if res.get("resolved_from"):
	out["resolved"] = {"from": res["resolved_from"], "name": res.get("name")}
	return out
	except ValueError as exc:
	raise HTTPException(status_code=400, detail=str(exc))


	@app.get("/api/track-record")
	def track_record(min_age_days: int = 30):
	"""Realized accuracy of past calls (direction hit-rate, mean abs error) once they age."""
	return analyst.track_record(min_age_days=min_age_days)


	@app.get("/api/recent")
	def recent():
	"""Cached analyses from the last 7 days — powers the Recent chips on the landing page."""
	return {"recent": analyst.recent_reports()}


	@app.get("/api/report/{ticker}/md")
	def cached_report_md(ticker: str):
	"""The stored audited markdown: the same fragments as the web page, byte for byte."""
	from fastapi.responses import PlainTextResponse
	md = analyst.cached_markdown(ticker)
	if not md:
	raise HTTPException(status_code=404, detail="no published analysis for this ticker")
	return PlainTextResponse(md, media_type="text/markdown; charset=utf-8",
	headers={"Content-Disposition": f'attachment; filename="Crucible_{ticker.upper()}.md"'})


	@app.get("/api/report/{ticker}")
	def cached_report(ticker: str):
	"""Serve a cached full analysis instantly (0 LLM calls, no engine lock). 404 when absent
	or older than 7 days; includes a staleness flag when the price has drifted >7% since."""
	d = analyst.cached_report(ticker)
	if not d:
	raise HTTPException(status_code=404, detail="no published analysis for this ticker")
	return d


	@app.get("/api/dev/qa/{ticker}")
	def dev_qa(ticker: str, x_dev_token: str = Header(default="")):
	"""Internal QA record (validation results, rejected sentences, state snapshot). Developer-only:
	requires the DEV_TOKEN secret; without it configured the route does not exist."""
	tok = os.getenv("DEV_TOKEN", "")
	if not tok:
	raise HTTPException(status_code=404, detail="not found")
	import hmac
	if not hmac.compare_digest(x_dev_token, tok):
	raise HTTPException(status_code=403, detail="forbidden")
	rec = analyst.qa_record(ticker)
	if rec is None:
	raise HTTPException(status_code=404, detail="no QA record for this ticker")
	return rec


	@app.post("/api/prewarm")
	def prewarm(x_prewarm_token: str = Header(default="")):
	"""Weekly, gated cache-warmer: only fires when organic searches haven't already filled
	the Recent row (>=3 fresh entries -> no-op). Token-gated so public callers can't spend
	LLM budget. Picks news-active large-caps not already cached, max 3."""
	import os as _o
	tok = _o.getenv("PREWARM_TOKEN", "")
	if not tok or x_prewarm_token != tok:
	raise HTTPException(status_code=403, detail="bad or missing prewarm token")
	fresh = analyst.recent_reports(limit=50)
	if len(fresh) >= 3:
	return {"started": [], "reason": f"organic cache sufficient ({len(fresh)} fresh entries)"}
	need = 3 - len(fresh)
	have = {r["ticker"] for r in fresh}
	started = []
	try:
	import urllib.request as _u, json as _j
	av = _o.getenv("ALPHAVANTAGE_API_KEY") or _o.getenv("ALPHA_VANTAGE_API_KEY", "")
	# What's trending IN THE NEWS: aggregate relevance-weighted ticker mentions across
	# the latest news feed (not "most traded", which skews to meme/penny volume).
	with _u.urlopen("https://www.alphavantage.co/query?function=NEWS_SENTIMENT"
	f"&sort=LATEST&limit=200&apikey={av}", timeout=25) as r:
	feed = _j.loads(r.read()).get("feed") or []
	score: dict = {}
	for a in feed:
	for ts in a.get("ticker_sentiment") or []:
	sym = (ts.get("ticker") or "").upper()
	if ":" in sym or not sym.isalpha() or len(sym) > 5: # skip CRYPTO:/FOREX:/funds
	continue
	score[sym] = score.get(sym, 0.0) + float(ts.get("relevance_score") or 0)
	for sym, _sc in sorted(score.items(), key=lambda x: -x[1]):
	if len(started) >= need:
	break
	if sym in have:
	continue
	try:
	res = analyst.resolve_symbol(sym) # verifies a real quote exists
	jid = analyst.start(res["symbol"])
	analyst._jobs[jid]["source"] = "prewarmed"
	started.append(res["symbol"]); have.add(res["symbol"])
	except ValueError:
	continue
	except Exception as exc:
	return {"started": started, "error": str(exc)[:120]}
	return {"started": started, "basis": "news-trending (relevance-weighted mentions, latest feed)"}


	@app.get("/api/jobs")
	def jobs():
	"""Who holds the engine, who is queued, recent jobs — for diagnosing 'stuck' runs."""
	return analyst.jobs_view()


	@app.get("/api/usage")
	def usage():
	"""Today's run count vs the daily cap (no side effects)."""
	today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
	used = 0
	try:
	with open(_CAP_FILE) as f:
	d = json.load(f)
	if d.get("date") == today:
	used = d.get("count", 0)
	except Exception:
	pass
	return {"date": today, "used": used, "cap": DAILY_CAP}


	@app.get("/api/analyst/{job_id}/stream")
	def analyst_stream(job_id: str):
	if not analyst.exists(job_id):
	raise HTTPException(status_code=404, detail="unknown job")
	return StreamingResponse(analyst.stream(job_id), media_type="text/event-stream")


	@app.get("/api/analyst/{job_id}/result")
	def analyst_result(job_id: str):
	if not analyst.exists(job_id):
	raise HTTPException(status_code=404, detail="unknown job")
	return analyst.result(job_id)


	@app.get("/")
	def index():
	return FileResponse(config.APP_DIR / "static" / "index.html", headers={"Cache-Control": "no-cache, must-revalidate"})


	app.mount("/static", StaticFiles(directory=config.APP_DIR / "static"), name="static")