From 3e1324eff69e4b121f85223758b872c7fb5dc027 Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 29 Mar 2026 15:41:42 -0700 Subject: Migrate insiders and filings from FMP to yfinance FMP v3 insider-trading and sec_filings endpoints are legacy-gated. Switch to yfinance (t.insider_transactions, t.sec_filings) which provides the same data for free with no API key required. Co-Authored-By: Claude Sonnet 4.6 --- components/filings.py | 88 +++++++++-------------- components/insiders.py | 190 ++++++++++++++++++++++--------------------------- 2 files changed, 119 insertions(+), 159 deletions(-) (limited to 'components') diff --git a/components/filings.py b/components/filings.py index d366a2b..a1e4417 100644 --- a/components/filings.py +++ b/components/filings.py @@ -1,8 +1,7 @@ """SEC filings — recent 10-K, 10-Q, 8-K and other forms with direct links.""" -import pandas as pd import streamlit as st -from services.fmp_service import get_sec_filings +from services.data_service import get_sec_filings _FORM_DESCRIPTIONS = { @@ -24,22 +23,19 @@ _FORM_COLORS = { } -def _describe_form(form_type: str) -> str: - return _FORM_DESCRIPTIONS.get(form_type.strip().upper(), "") - - def render_filings(ticker: str): with st.spinner("Loading SEC filings…"): - raw = get_sec_filings(ticker) + filings = get_sec_filings(ticker) - if not raw: - st.info("No SEC filing data available. Requires FMP API key.") + if not filings: + st.info("No SEC filing data available for this ticker.") return - # Collect unique form types for filter - form_types = sorted({str(r.get("type") or r.get("form") or "").strip() for r in raw if r.get("type") or r.get("form")}) - filter_options = ["All"] + [t for t in ["10-K", "10-Q", "8-K"] if t in form_types] + \ - [t for t in form_types if t not in ("10-K", "10-Q", "8-K")] + # yfinance returns: date (datetime.date), type, title, edgarUrl, exhibits (dict) + form_types = sorted({str(f.get("type", "")).strip() for f in filings if f.get("type")}) + priority = [t for t in ["10-K", "10-Q", "8-K"] if t in form_types] + other = [t for t in form_types if t not in ("10-K", "10-Q", "8-K")] + filter_options = ["All"] + priority + other filter_col, _ = st.columns([1, 3]) with filter_col: @@ -47,69 +43,49 @@ def render_filings(ticker: str): # Summary counts counts = {} - for r in raw: - ft = str(r.get("type") or r.get("form") or "Other").strip() + for f in filings: + ft = str(f.get("type", "Other")).strip() counts[ft] = counts.get(ft, 0) + 1 - priority_forms = ["10-K", "10-Q", "8-K"] - summary_forms = [f for f in priority_forms if f in counts] - if summary_forms: - cols = st.columns(len(summary_forms)) - for col, ft in zip(cols, summary_forms): - col.metric(ft, counts[ft]) + if priority: + cols = st.columns(len(priority)) + for col, ft in zip(cols, priority): + col.metric(ft, counts.get(ft, 0)) st.write("") - # Filter - if selected_type == "All": - filtered = raw - else: - filtered = [ - r for r in raw - if str(r.get("type") or r.get("form") or "").strip() == selected_type - ] + filtered = filings if selected_type == "All" else [ + f for f in filings if str(f.get("type", "")).strip() == selected_type + ] if not filtered: st.info("No filings match the current filter.") return - # Build display table - rows = [] - for item in filtered: - form_type = str(item.get("type") or item.get("form") or "—").strip() - date = str(item.get("filingDate") or item.get("date") or "")[:10] - description = item.get("description") or _describe_form(form_type) or "—" - url = item.get("link") or item.get("finalLink") or item.get("url") or "" - rows.append({ - "Date": date, - "Form": form_type, - "Description": description, - "Link": url, - }) - - df = pd.DataFrame(rows) - - # Render as clickable table — Streamlit doesn't natively support link columns, - # so we render each row as a compact card for the most recent 30 entries. - for _, row in df.head(30).iterrows(): - form = row["Form"] + for item in filtered[:40]: + form = str(item.get("type", "—")).strip() + date = str(item.get("date", ""))[:10] + title = item.get("title") or _FORM_DESCRIPTIONS.get(form, "") + edgar_url = item.get("edgarUrl", "") + exhibits = item.get("exhibits") or {} + color = _FORM_COLORS.get(form, "rgba(255,255,255,0.05)") - date = row["Date"] - desc = row["Description"] - link = row["Link"] with st.container(): left, right = st.columns([5, 1]) with left: st.markdown( f"
" - f"{form}  ·  {date}
" - f"{desc}" + f"{form}  ·  " + f"{date}
" + f"{title}" f"
", unsafe_allow_html=True, ) with right: - if link: + # Prefer the actual filing doc over the Yahoo index page + doc_url = exhibits.get(form) or edgar_url + if doc_url: st.markdown( - f"
🔗 View
", + f"
🔗 View
", unsafe_allow_html=True, ) diff --git a/components/insiders.py b/components/insiders.py index 05561ba..354ffef 100644 --- a/components/insiders.py +++ b/components/insiders.py @@ -4,121 +4,102 @@ import plotly.graph_objects as go import streamlit as st from datetime import datetime, timedelta -from services.fmp_service import get_insider_transactions +from services.data_service import get_insider_transactions from utils.formatters import fmt_currency, fmt_large -def _parse_date(date_str: str) -> datetime | None: - for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"): - try: - return datetime.strptime(str(date_str)[:19], fmt) - except ValueError: - continue - return None - - -def _classify(row: dict) -> str: - """Return 'Buy' or 'Sell' from the transaction type field.""" - tx = str(row.get("transactionType") or row.get("acquistionOrDisposition") or "").strip() - buy_keywords = ("purchase", "p -", "p-", "acquisition", "a -", "a-", "grant", "award", "exercise") - sell_keywords = ("sale", "s -", "s-", "disposition", "d -", "d-") - tx_lower = tx.lower() - if any(k in tx_lower for k in sell_keywords): - return "Sell" - if any(k in tx_lower for k in buy_keywords): - return "Buy" - # Fallback: FMP often uses acquistionOrDisposition = "A" or "D" - aod = str(row.get("acquistionOrDisposition") or "").strip().upper() - if aod == "D": +def _classify(text: str) -> str: + """Return 'Buy', 'Sell', or 'Other' from the transaction text.""" + t = str(text or "").lower() + if any(k in t for k in ("sale", "sold", "disposition")): return "Sell" - if aod == "A": + if any(k in t for k in ("purchase", "bought", "acquisition", "grant", "award", "exercise")): return "Buy" return "Other" def render_insiders(ticker: str): with st.spinner("Loading insider transactions…"): - raw = get_insider_transactions(ticker) + df = get_insider_transactions(ticker) - if not raw: - st.info("No insider transaction data available. Requires FMP API key.") + if df.empty: + st.info("No insider transaction data available for this ticker.") return - # Build enriched rows - rows = [] - for item in raw: - direction = _classify(item) - shares = item.get("securitiesTransacted") or item.get("shares") or 0 - price = item.get("price") or 0.0 + # Normalise columns — yfinance returns: Shares, URL, Text, Insider, Position, + # Transaction, Start Date, Ownership, Value + df = df.copy() + df["Direction"] = df["Text"].apply(_classify) + + # Parse dates + def _to_dt(val): try: - value = float(shares) * float(price) - except (TypeError, ValueError): - value = 0.0 - date_obj = _parse_date(str(item.get("transactionDate") or item.get("filingDate") or "")) - rows.append({ - "date": date_obj, - "date_str": str(item.get("transactionDate") or "")[:10], - "name": item.get("reportingName") or item.get("insiderName") or "—", - "title": item.get("typeOfOwner") or item.get("title") or "—", - "direction": direction, - "shares": shares, - "price": price, - "value": value, - "filing_url": item.get("link") or item.get("secLink") or "", - }) + return pd.to_datetime(val) + except Exception: + return pd.NaT + + df["_date"] = df["Start Date"].apply(_to_dt) # ── Summary: last 6 months ──────────────────────────────────────────── - cutoff = datetime.now() - timedelta(days=180) - recent = [r for r in rows if r["date"] and r["date"] >= cutoff] + cutoff = pd.Timestamp(datetime.now() - timedelta(days=180)) + recent = df[df["_date"] >= cutoff] - buys = [r for r in recent if r["direction"] == "Buy"] - sells = [r for r in recent if r["direction"] == "Sell"] - total_buy_val = sum(r["value"] for r in buys) - total_sell_val = sum(r["value"] for r in sells) + buys = recent[recent["Direction"] == "Buy"] + sells = recent[recent["Direction"] == "Sell"] + + def _total_value(subset): + try: + return subset["Value"].dropna().astype(float).sum() + except Exception: + return 0.0 + + buy_val = _total_value(buys) + sell_val = _total_value(sells) st.markdown("**Insider Activity — Last 6 Months**") c1, c2, c3, c4 = st.columns(4) c1.metric("Buy Transactions", len(buys)) - c2.metric("Total Bought", fmt_large(total_buy_val) if total_buy_val else "—") + c2.metric("Total Bought", fmt_large(buy_val) if buy_val else "—") c3.metric("Sell Transactions", len(sells)) - c4.metric("Total Sold", fmt_large(total_sell_val) if total_sell_val else "—") + c4.metric("Total Sold", fmt_large(sell_val) if sell_val else "—") - # Net buy/sell bar chart (monthly) - if recent: + # Monthly bar chart + if not recent.empty: monthly: dict[str, dict] = {} - for r in recent: - if r["date"]: - key = r["date"].strftime("%Y-%m") - monthly.setdefault(key, {"Buy": 0.0, "Sell": 0.0}) - if r["direction"] in ("Buy", "Sell"): - monthly[key][r["direction"]] += r["value"] + for _, row in recent.iterrows(): + if pd.isna(row["_date"]): + continue + key = row["_date"].strftime("%Y-%m") + monthly.setdefault(key, {"Buy": 0.0, "Sell": 0.0}) + try: + val = float(row["Value"]) if pd.notna(row["Value"]) else 0.0 + except (TypeError, ValueError): + val = 0.0 + if row["Direction"] in ("Buy", "Sell"): + monthly[key][row["Direction"]] += val months = sorted(monthly.keys()) - buy_vals = [monthly[m]["Buy"] / 1e6 for m in months] - sell_vals = [-monthly[m]["Sell"] / 1e6 for m in months] - - fig = go.Figure() - fig.add_trace(go.Bar( - x=months, y=buy_vals, - name="Buys", - marker_color="#2ecc71", - )) - fig.add_trace(go.Bar( - x=months, y=sell_vals, - name="Sells", - marker_color="#e74c3c", - )) - fig.update_layout( - title="Monthly Insider Net Activity ($M)", - barmode="relative", - yaxis_title="Value ($M)", - plot_bgcolor="rgba(0,0,0,0)", - paper_bgcolor="rgba(0,0,0,0)", - margin=dict(l=0, r=0, t=40, b=0), - height=280, - legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1), - ) - st.plotly_chart(fig, use_container_width=True) + if months: + fig = go.Figure() + fig.add_trace(go.Bar( + x=months, y=[monthly[m]["Buy"] / 1e6 for m in months], + name="Buys", marker_color="#2ecc71", + )) + fig.add_trace(go.Bar( + x=months, y=[-monthly[m]["Sell"] / 1e6 for m in months], + name="Sells", marker_color="#e74c3c", + )) + fig.update_layout( + title="Monthly Insider Net Activity ($M)", + barmode="relative", + yaxis_title="Value ($M)", + plot_bgcolor="rgba(0,0,0,0)", + paper_bgcolor="rgba(0,0,0,0)", + margin=dict(l=0, r=0, t=40, b=0), + height=280, + legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1), + ) + st.plotly_chart(fig, use_container_width=True) st.divider() @@ -131,31 +112,34 @@ def render_insiders(ticker: str): "Filter", options=["All", "Buy", "Sell"], index=0, key="insider_filter" ) - filtered = rows if direction_filter == "All" else [r for r in rows if r["direction"] == direction_filter] + filtered = df if direction_filter == "All" else df[df["Direction"] == direction_filter] - if not filtered: + if filtered.empty: st.info("No transactions match the current filter.") return - df = pd.DataFrame([{ - "Date": r["date_str"], - "Insider": r["name"], - "Title": r["title"], - "Type": r["direction"], - "Shares": f"{int(r['shares']):,}" if r["shares"] else "—", - "Price": fmt_currency(r["price"]) if r["price"] else "—", - "Value": fmt_large(r["value"]) if r["value"] else "—", - } for r in filtered]) + display = pd.DataFrame({ + "Date": filtered["Start Date"].astype(str).str[:10], + "Insider": filtered["Insider"], + "Position": filtered["Position"], + "Type": filtered["Direction"], + "Shares": filtered["Shares"].apply( + lambda v: f"{int(v):,}" if pd.notna(v) else "—" + ), + "Value": filtered["Value"].apply( + lambda v: fmt_large(float(v)) if pd.notna(v) and float(v) > 0 else "—" + ), + }).reset_index(drop=True) def _color_type(row): if row["Type"] == "Buy": - return [""] * 2 + ["background-color: rgba(46,204,113,0.15)"] + [""] * 4 + return [""] * 3 + ["background-color: rgba(46,204,113,0.15)"] + [""] * 2 if row["Type"] == "Sell": - return [""] * 2 + ["background-color: rgba(231,76,60,0.15)"] + [""] * 4 + return [""] * 3 + ["background-color: rgba(231,76,60,0.15)"] + [""] * 2 return [""] * len(row) st.dataframe( - df.style.apply(_color_type, axis=1), + display.style.apply(_color_type, axis=1), use_container_width=True, hide_index=True, ) -- cgit v1.3-2-g0d8e