""" signals.py — C25 Signal Board & Reporting Usage: python3 signals.py # full signal board (last 7 days) python3 signals.py top [--days 14] # top companies by mentions python3 signals.py company NOVO-B # detail view for one company python3 signals.py summary # quick one-liner per company python3 signals.py buy # only budget/accessible + leveraged ETPs """ import sys import json import time import argparse import sqlite3 from datetime import datetime, timezone from pathlib import Path import yfinance as yf from db import get_conn, DBConn C25_PATH = Path(__file__).parent / "c25.json" _c25_raw = json.loads(C25_PATH.read_text()) C25: dict[str, dict] = {k: v for k, v in _c25_raw.items() if not k.startswith("_")} _analyst_cache: dict[str, dict] = {} def get_db() -> DBConn: """Return a DBConn wrapper. Schema is managed by db.py.""" return get_conn() def analyst_rec(ticker: str) -> dict: """Hent analytikernes konsensus fra Yahoo Finance (cachet pr. kørsel).""" if ticker in _analyst_cache: return _analyst_cache[ticker] yf_ticker = C25.get(ticker, {}).get("ticker_yahoo", ticker + ".CO") try: info = yf.Ticker(yf_ticker).info mean = info.get("recommendationMean") count = info.get("numberOfAnalystOpinions", 0) if mean is None: result = {"label": "Ukendt", "mean": None, "count": 0} elif mean <= 1.5: result = {"label": "STÆRK KØB 🟢", "mean": mean, "count": count} elif mean <= 2.5: result = {"label": "KØB 🟢", "mean": mean, "count": count} elif mean <= 3.5: result = {"label": "HOLD 🟡", "mean": mean, "count": count} elif mean <= 4.5: result = {"label": "SÆLG 🔴", "mean": mean, "count": count} else: result = {"label": "STÆRK SÆLG 🔴", "mean": mean, "count": count} except Exception: result = {"label": "Ukendt", "mean": None, "count": 0} _analyst_cache[ticker] = result return result SENTIMENT_DA = {"positive": "POSITIV ↑", "negative": "NEGATIV ↓", "neutral": "NEUTRAL →"} TIER_DA = { "budget": "under 200 kr — kan købes direkte", "accessible": "200-500 kr — kan købes direkte", "expensive": "500-2000 kr — overvej ETP", "inaccessible": "over 2000 kr — brug ETP", } # --------------------------------------------------------------------------- # Queries # --------------------------------------------------------------------------- def company_stats(db: DBConn, days: int = 7) -> list[dict]: """Return per-company aggregated signal stats for the last N days.""" cutoff = int(time.time()) - days * 86400 rows = db.execute( """SELECT s.ticker, COUNT(*) AS mention_articles, SUM(s.mention_count) AS total_mentions, AVG(CASE WHEN s.sentiment='positive' THEN s.sentiment_score WHEN s.sentiment='negative' THEN -s.sentiment_score ELSE 0 END) AS avg_sentiment, SUM(CASE WHEN s.sentiment='positive' THEN 1 ELSE 0 END) AS pos_count, SUM(CASE WHEN s.sentiment='negative' THEN 1 ELSE 0 END) AS neg_count, SUM(CASE WHEN s.sentiment='neutral' THEN 1 ELSE 0 END) AS neu_count, AVG(a.source_count) AS avg_sources, MAX(COALESCE(s.signal_score, 0)) AS max_signal, SUM(COALESCE(s.alert, 0)) AS alert_count FROM article_signals s JOIN articles a ON a.slug = s.article_slug WHERE s.analyzed_at >= ? GROUP BY s.ticker ORDER BY max_signal DESC, mention_articles DESC""", (cutoff,), ).fetchall() return [dict(r) for r in rows] def company_articles(db: DBConn, ticker: str, days: int = 14) -> list[dict]: """Return individual articles mentioning a company.""" cutoff = int(time.time()) - days * 86400 rows = db.execute( """SELECT a.title, a.slug, a.start_date, a.source_count, s.sentiment, s.sentiment_score, s.mention_count, s.signal_score, s.alert, s.claude_reasoning, s.momentum_dir, s.momentum_pct_5d, s.coverage_spread FROM article_signals s JOIN articles a ON a.slug = s.article_slug WHERE s.ticker = ? AND s.analyzed_at >= ? ORDER BY COALESCE(s.signal_score,0) DESC, a.source_count DESC""", (ticker, cutoff), ).fetchall() return [dict(r) for r in rows] # --------------------------------------------------------------------------- # Display helpers # --------------------------------------------------------------------------- def print_signal_board(days: int = 7) -> None: db = get_db() rows = company_stats(db, days) db.close() if not rows: print(f"Ingen signaler de seneste {days} dage. Kør: make") return now_str = datetime.now(timezone.utc).strftime("%d %b %Y %H:%M") W = 66 print(f"\n{'═'*W}") print(f" MONEYMAKER · Hvad sker der med dine C25 aktier? · {now_str}") print(f"{'═'*W}") print(f" (baseret på de seneste {days} dage med nyheder)\n") for r in rows: ticker = r["ticker"] company = C25.get(ticker, {}) name = company.get("name", ticker) price = company.get("price_dkk_approx", "?") tier = company.get("tier", "?") etps = company.get("leveraged", []) avg_s = r["avg_sentiment"] or 0 sent_da = SENTIMENT_DA.get( "positive" if avg_s > 0.1 else ("negative" if avg_s < -0.1 else "neutral"), "NEUTRAL →" ) n_art = r["mention_articles"] n_src = int(r["avg_sources"] or 0) sig = r.get("max_signal") or 0 alerts = r.get("alert_count") or 0 # Analytiker konsensus rec = analyst_rec(ticker) # Er nyheder og analytikere enige? news_bull = avg_s > 0.1 news_bear = avg_s < -0.1 ana_bull = rec["mean"] is not None and rec["mean"] <= 2.5 ana_bear = rec["mean"] is not None and rec["mean"] >= 3.5 if (news_bull and ana_bull) or (news_bear and ana_bear): agreement = "✅ Nyheder og eksperter er ENIGE" elif (news_bull and ana_bear) or (news_bear and ana_bull): agreement = "⚠️ Nyheder og eksperter er UENIGE" else: agreement = "〰️ Intet klart signal endnu" # Kildedækning advarsel coverage_warn = " ⚠️ Kun få mediekilder — usikkert" if n_src < 3 else "" # Signal i ord if sig >= 0.5: sig_txt = f"🔥 STÆRKT ({sig:.2f})" elif sig >= 0.2: sig_txt = f"📊 Moderat ({sig:.2f})" elif sig > 0: sig_txt = f"🔍 Svagt ({sig:.2f})" else: sig_txt = f"⬜ Ingen ({sig:.2f})" # ETP info etp_info = "" if etps and len(etps) >= 2: long_e = next((e for e in etps if e["direction"] == "long"), etps[0]) short_e = next((e for e in etps if e["direction"] == "short"), etps[1]) etp_info = f"\n │ ETP (gearing 3x): KØB={long_e['ticker']} SÆLG={short_e['ticker']} ({long_e['exchange']})" alerts_str = f"\n │ ⚡ {alerts} alert(s) udløst!" if alerts else "" print(f" ┌─ {name} ({ticker}) {'─'*max(1, W-6-len(name)-len(ticker))}") print(f" │ Pris: ca. {price} kr · {TIER_DA.get(tier, tier)}") print(f" │ Eksperter ({rec['count']:>2}): {rec['label']}") print(f" │ Nyheder ({n_art} art): {sent_da} · gnsn. {n_src} mediekilder{coverage_warn}") print(f" │ Samlet vurdering: {agreement}") print(f" │ Signalstyrke: {sig_txt}{etp_info}{alerts_str}") print(f" └{'─'*W}") print() print(f" Forklaring:") print(f" Eksperter = finansanalytikere der følger aktien (Yahoo Finance)") print(f" Nyheder = hvad medierne skriver, analyseret med AI (FinBERT + Claude)") print(f" Signal = kombineret score: nyhedssentiment × kildedækning × kursmomentum") print(f" ETP = børshandlet certifikat — du køber 3x gearing uden at eje aktien") print() def print_company_detail(ticker: str, days: int = 14) -> None: ticker = ticker.upper() if ticker not in C25: print(f"Ukendt ticker: {ticker}. Tilgængelige: {', '.join(sorted(C25.keys()))}") return db = get_db() company = C25[ticker] stats = next((r for r in company_stats(db, days) if r["ticker"] == ticker), None) arts = company_articles(db, ticker, days) db.close() rec = analyst_rec(ticker) price = company.get("price_dkk_approx", "?") tier = company.get("tier", "?") etps = company.get("leveraged", []) W = 66 print(f"\n{'═'*W}") print(f" {company['name']} ({ticker})") print(f" Sektor: {company['sector']} · ca. {price} kr · {TIER_DA.get(tier, tier)}") if etps: for e in etps: label = "KØB (long)" if e["direction"] == "long" else "SÆLG (short)" print(f" ETP {label}: {e['name']} ({e['ticker']}) — {e['exchange']}") print(f"{'─'*W}") print(f"\n Hvad siger eksperterne?") print(f" {rec['label']} · {rec['count']} analytikere følger aktien") print(f"\n Hvad siger nyhederne de seneste {days} dage?") if not stats: print(f" Ingen nyheder fundet endnu.") return avg_s = stats["avg_sentiment"] or 0 sent_da = SENTIMENT_DA.get( "positive" if avg_s > 0.1 else ("negative" if avg_s < -0.1 else "neutral"), "" ) print(f" {sent_da} · {stats['mention_articles']} artikler") print(f" Positiv: {stats['pos_count']} · Negativ: {stats['neg_count']} · Neutral: {stats['neu_count']}") print(f"\n Enkeltartikler:") print(f" {'─'*W}") for a in arts[:15]: sent_da2 = SENTIMENT_DA.get(a["sentiment"], "?") sig = a.get("signal_score") or 0 n_src = a.get("source_count") or 0 low_cov = " ⚠️ få kilder" if n_src < 3 else "" sig_txt = f"🔥{sig:.2f}" if sig >= 0.5 else (f"📊{sig:.2f}" if sig >= 0.2 else f"⬜{sig:.2f}") print(f"\n {a['title'][:62]}") print(f" Sentiment: {sent_da2:<14} Signal: {sig_txt} Kilder: {n_src}{low_cov}") reason = a.get("claude_reasoning", "") if reason and reason not in ("(no API key)", ""): print(f" Claude: {reason[:80]}") def print_buy_signals(days: int = 7) -> None: """Vis kun aktier med positivt signal — direkte køb og via ETP.""" db = get_db() rows = company_stats(db, days) db.close() W = 66 print(f"\n{'═'*W}") print(f" HVAD KAN DET BETALE SIG? · de seneste {days} dage") print(f"{'═'*W}\n") direct = [r for r in rows if C25.get(r["ticker"], {}).get("tier") in ("budget", "accessible")] via_etp = [r for r in rows if C25.get(r["ticker"], {}).get("leveraged")] def fmt(r: dict, note: str = "") -> None: ticker = r["ticker"] c = C25.get(ticker, {}) avg_s = r["avg_sentiment"] or 0 sent = SENTIMENT_DA.get( "positive" if avg_s > 0.1 else ("negative" if avg_s < -0.1 else "neutral"), "") rec = analyst_rec(ticker) price = c.get("price_dkk_approx", "?") sig = r.get("max_signal") or 0 print(f" {c.get('name',''):<30} {price:>6} kr") print(f" Eksperter: {rec['label']}") print(f" Nyheder: {sent} (signal: {sig:.2f})") if note: print(f" {note}") print() if direct: print(" Du kan købe disse DIREKTE (overkommelig pris):") print(f" {'─'*W}") for r in sorted(direct, key=lambda x: abs(x["avg_sentiment"] or 0), reverse=True): fmt(r) if via_etp: print(" Via WisdomTree ETP (3x gearing — højere risiko/gevinst):") print(f" {'─'*W}") for r in sorted(via_etp, key=lambda x: abs(x["avg_sentiment"] or 0), reverse=True): ticker = r["ticker"] avg_s = r["avg_sentiment"] or 0 etps = C25.get(ticker, {}).get("leveraged", []) if etps: direction = "long" if avg_s >= 0 else "short" etp = next((e for e in etps if e["direction"] == direction), etps[0]) fmt(r, note=f"ETP: {etp['ticker']} ({direction} 3x) på {etp['exchange']}") else: fmt(r) if not direct and not via_etp: print(" Ingen signaler endnu. Kør: make") # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main() -> None: parser = argparse.ArgumentParser(description="C25 signal board") sub = parser.add_subparsers(dest="cmd") p_top = sub.add_parser("top", help="Top companies by mention count") p_top.add_argument("--days", type=int, default=7) p_co = sub.add_parser("company", help="Detail view for one company") p_co.add_argument("ticker") p_co.add_argument("--days", type=int, default=14) sub.add_parser("summary", help="One-liner per company") p_buy = sub.add_parser("buy", help="Affordable + leveraged ETP candidates") p_buy.add_argument("--days", type=int, default=7) p_board = sub.add_parser("board", help="Full signal board") p_board.add_argument("--days", type=int, default=7) args = parser.parse_args() if args.cmd in (None, "board"): days = getattr(args, "days", 7) print_signal_board(days) elif args.cmd == "top": print_signal_board(args.days) elif args.cmd == "company": print_company_detail(args.ticker, getattr(args, "days", 14)) elif args.cmd == "buy": print_buy_signals(args.days) elif args.cmd == "summary": print_signal_board(7) if __name__ == "__main__": main()