"""Insider transactions — recent buys/sells with summary and detail table.""" import pandas as pd import plotly.graph_objects as go import streamlit as st from datetime import datetime, timedelta from services.fmp_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": return "Sell" if aod == "A": return "Buy" return "Other" def render_insiders(ticker: str): with st.spinner("Loading insider transactions…"): raw = get_insider_transactions(ticker) if not raw: st.info("No insider transaction data available. Requires FMP API key.") 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 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 "", }) # ── Summary: last 6 months ──────────────────────────────────────────── cutoff = datetime.now() - timedelta(days=180) recent = [r for r in rows if r["date"] and r["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) 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 "—") c3.metric("Sell Transactions", len(sells)) c4.metric("Total Sold", fmt_large(total_sell_val) if total_sell_val else "—") # Net buy/sell bar chart (monthly) if recent: 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"] 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) st.divider() # ── Transaction table ───────────────────────────────────────────────── st.markdown("**All Transactions**") filter_col, _ = st.columns([1, 3]) with filter_col: direction_filter = st.selectbox( "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] if not filtered: 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]) def _color_type(row): if row["Type"] == "Buy": return [""] * 2 + ["background-color: rgba(46,204,113,0.15)"] + [""] * 4 if row["Type"] == "Sell": return [""] * 2 + ["background-color: rgba(231,76,60,0.15)"] + [""] * 4 return [""] * len(row) st.dataframe( df.style.apply(_color_type, axis=1), use_container_width=True, hide_index=True, )