import asyncio import json import time from datetime import datetime, timezone from pathlib import Path import httpx from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles APP_DIR = Path(__file__).resolve().parent SEC_AGENT = "Robinhood open research prototype contact: https://huggingface.co/pumpfun/Robinhood" SEC_HEADERS = {"User-Agent": SEC_AGENT, "Accept": "application/json"} TICKERS_URL = "https://www.sec.gov/files/company_tickers.json" SUBMISSIONS_URL = "https://data.sec.gov/submissions/CIK{cik}.json" FACTS_URL = "https://data.sec.gov/api/xbrl/companyfacts/CIK{cik}.json" FORMS = {"10-K", "10-K/A", "10-Q", "10-Q/A", "8-K", "8-K/A", "20-F", "20-F/A", "6-K", "6-K/A"} METRICS = [ ("revenue", "Revenue", "营业收入", ["RevenueFromContractWithCustomerExcludingAssessedTax", "Revenues", "SalesRevenueNet"]), ("net_income", "Net income", "净利润", ["NetIncomeLoss", "ProfitLoss"]), ("assets", "Total assets", "总资产", ["Assets"]), ("liabilities", "Total liabilities", "总负债", ["Liabilities"]), ("cash", "Cash & equivalents", "现金及等价物", ["CashAndCashEquivalentsAtCarryingValue", "CashCashEquivalentsRestrictedCashAndRestrictedCashEquivalents"]), ] app = FastAPI(title="Robinhood public company research", docs_url=None, redoc_url=None) app.add_middleware( CORSMiddleware, allow_origins=["https://pumpfun-robinhood-web.static.hf.space"], allow_methods=["GET"], allow_headers=["*"], ) _ticker_cache = {"expires": 0.0, "rows": []} def fetch_json(url: str) -> dict: try: response = httpx.get(url, headers=SEC_HEADERS, timeout=15, follow_redirects=True) response.raise_for_status() return response.json() except httpx.HTTPStatusError as exc: raise RuntimeError(f"SEC returned HTTP {exc.response.status_code}") from exc except (httpx.HTTPError, json.JSONDecodeError) as exc: raise RuntimeError("SEC data is temporarily unavailable") from exc def ticker_rows() -> list[dict]: now = time.time() if _ticker_cache["rows"] and _ticker_cache["expires"] > now: return _ticker_cache["rows"] payload = fetch_json(TICKERS_URL) rows = list(payload.values()) _ticker_cache.update(rows=rows, expires=now + 21600) return rows def resolve_company(query: str) -> dict | None: needle = query.strip().upper() rows = ticker_rows() exact_ticker = next((row for row in rows if row["ticker"].upper() == needle), None) if exact_ticker: return exact_ticker exact_name = next((row for row in rows if row["title"].upper() == needle), None) if exact_name: return exact_name starts = [row for row in rows if row["title"].upper().startswith(needle)] if starts: return min(starts, key=lambda row: len(row["title"])) contains = [row for row in rows if needle in row["title"].upper()] return min(contains, key=lambda row: len(row["title"])) if contains else None def filing_rows(submissions: dict) -> list[dict]: recent = submissions.get("filings", {}).get("recent", {}) keys = ("accessionNumber", "filingDate", "reportDate", "form", "primaryDocument", "primaryDocDescription") count = len(recent.get("form", [])) filings = [] cik_plain = str(int(submissions["cik"])) for index in range(count): row = {key: recent.get(key, [""] * count)[index] for key in keys} if row["form"] not in FORMS or not row["primaryDocument"]: continue accession_plain = row["accessionNumber"].replace("-", "") filings.append({ "form": row["form"], "filed": row["filingDate"], "period": row["reportDate"], "description": row["primaryDocDescription"] or row["form"], "url": f"https://www.sec.gov/Archives/edgar/data/{cik_plain}/{accession_plain}/{row['primaryDocument']}", }) if len(filings) == 8: break return filings def latest_metric(facts: dict, metric: tuple) -> dict | None: key, label_en, label_zh, tags = metric us_gaap = facts.get("facts", {}).get("us-gaap", {}) for tag in tags: concept = us_gaap.get(tag) if not concept: continue units = concept.get("units", {}) points = units.get("USD") or next(iter(units.values()), []) valid = [point for point in points if point.get("form") in FORMS and point.get("val") is not None and point.get("filed")] if not valid: continue point = max(valid, key=lambda item: (item.get("filed", ""), item.get("end", ""))) return { "key": key, "label_en": label_en, "label_zh": label_zh, "value": point["val"], "unit": "USD", "period_start": point.get("start"), "period": point.get("end"), "filed": point.get("filed"), "form": point.get("form"), "taxonomy_tag": tag, } return None def build_research(query: str) -> dict: match = resolve_company(query) if not match: raise HTTPException(status_code=404, detail="Company not found. Try a U.S. ticker such as AAPL or MSFT.") cik = str(match["cik_str"]).zfill(10) submissions = fetch_json(SUBMISSIONS_URL.format(cik=cik)) facts = fetch_json(FACTS_URL.format(cik=cik)) metrics = [result for metric in METRICS if (result := latest_metric(facts, metric))] return { "company": { "name": submissions.get("name") or match["title"], "ticker": (submissions.get("tickers") or [match["ticker"]])[0], "exchange": (submissions.get("exchanges") or [""])[0], "cik": cik, "sic": submissions.get("sic"), "sic_description": submissions.get("sicDescription"), "fiscal_year_end": submissions.get("fiscalYearEnd"), "state": submissions.get("stateOfIncorporation"), }, "metrics": metrics, "filings": filing_rows(submissions), "retrieved_at": datetime.now(timezone.utc).isoformat(), "source": "U.S. Securities and Exchange Commission EDGAR", "limitations": { "en": "Figures reflect the latest matching standardized XBRL facts, whose periods may differ. Coverage and tagging vary by issuer. Verify decisions in the linked filings.", "zh": "数字来自最近匹配到的标准化 XBRL 数据，各指标期间可能不同；不同公司的覆盖和标签质量也有差异。请在原始申报中核实。", }, } @app.get("/api/health") def health(): return {"status": "ok"} @app.get("/api/research") async def research(q: str = Query(min_length=1, max_length=80)): clean = " ".join(q.split()) try: return await asyncio.to_thread(build_research, clean) except HTTPException: raise except RuntimeError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc app.mount("/", StaticFiles(directory=APP_DIR, html=True), name="site")