355 lines
14 KiB
Python
355 lines
14 KiB
Python
"""
|
||
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()
|