447 lines
16 KiB
Python
447 lines
16 KiB
Python
"""
|
|
Report Generator for POTE
|
|
|
|
Generates formatted reports from database data.
|
|
"""
|
|
|
|
import logging
|
|
from datetime import date, datetime, timedelta
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from pote.db.models import MarketAlert, Official, Security, Trade
|
|
from pote.monitoring.disclosure_correlator import DisclosureCorrelator
|
|
from pote.monitoring.pattern_detector import PatternDetector
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ReportGenerator:
|
|
"""Generates various types of reports from database data."""
|
|
|
|
def __init__(self, session: Session):
|
|
self.session = session
|
|
self.correlator = DisclosureCorrelator(session)
|
|
self.detector = PatternDetector(session)
|
|
|
|
def generate_daily_summary(
|
|
self, report_date: Optional[date] = None, *, lookback_days: int = 1
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Generate a daily summary report.
|
|
|
|
Args:
|
|
report_date: Date to generate report for (defaults to today)
|
|
lookback_days: Include trades filed in the last N days ending on report_date
|
|
(defaults to 1, meaning only filings on report_date).
|
|
|
|
Returns:
|
|
Dictionary containing report data
|
|
"""
|
|
if report_date is None:
|
|
report_date = date.today()
|
|
|
|
if lookback_days < 1:
|
|
raise ValueError("lookback_days must be >= 1")
|
|
|
|
start_of_day = datetime.combine(report_date, datetime.min.time())
|
|
end_of_day = datetime.combine(report_date, datetime.max.time())
|
|
|
|
filing_start_date = report_date - timedelta(days=lookback_days - 1)
|
|
|
|
# Trades filed within the lookback window (inclusive)
|
|
new_trades = (
|
|
self.session.query(Trade)
|
|
.filter(Trade.filing_date >= filing_start_date, Trade.filing_date <= report_date)
|
|
.all()
|
|
)
|
|
|
|
# Count market alerts today
|
|
new_alerts = (
|
|
self.session.query(MarketAlert)
|
|
.filter(
|
|
MarketAlert.timestamp >= start_of_day,
|
|
MarketAlert.timestamp <= end_of_day,
|
|
)
|
|
.all()
|
|
)
|
|
|
|
# Get high-severity alerts
|
|
critical_alerts = [a for a in new_alerts if a.severity >= 7]
|
|
|
|
# Get suspicious timing matches
|
|
suspicious_trades = []
|
|
for trade in new_trades:
|
|
analysis = self.correlator.analyze_trade(trade)
|
|
if analysis["timing_score"] >= 50:
|
|
suspicious_trades.append(analysis)
|
|
|
|
return {
|
|
"date": report_date,
|
|
"filing_start_date": filing_start_date,
|
|
"lookback_days": lookback_days,
|
|
"new_trades_count": len(new_trades),
|
|
"new_trades": [
|
|
{
|
|
"official": t.official.name if t.official else "Unknown",
|
|
"ticker": t.security.ticker if t.security else "Unknown",
|
|
"side": t.side,
|
|
"transaction_date": t.transaction_date,
|
|
"value_min": t.value_min,
|
|
"value_max": t.value_max,
|
|
}
|
|
for t in new_trades
|
|
],
|
|
"market_alerts_count": len(new_alerts),
|
|
"critical_alerts_count": len(critical_alerts),
|
|
"critical_alerts": [
|
|
{
|
|
"ticker": a.ticker,
|
|
"type": a.alert_type,
|
|
"severity": a.severity,
|
|
"timestamp": a.timestamp,
|
|
"details": a.details,
|
|
}
|
|
for a in critical_alerts
|
|
],
|
|
"suspicious_trades_count": len(suspicious_trades),
|
|
"suspicious_trades": suspicious_trades,
|
|
}
|
|
|
|
def generate_weekly_summary(self) -> Dict[str, Any]:
|
|
"""
|
|
Generate a weekly summary report.
|
|
|
|
Returns:
|
|
Dictionary containing report data
|
|
"""
|
|
week_ago = date.today() - timedelta(days=7)
|
|
|
|
# Most active officials
|
|
active_officials = (
|
|
self.session.query(
|
|
Official.name, func.count(Trade.id).label("trade_count")
|
|
)
|
|
.join(Trade)
|
|
.filter(Trade.filing_date >= week_ago)
|
|
.group_by(Official.id, Official.name)
|
|
.order_by(func.count(Trade.id).desc())
|
|
.limit(10)
|
|
.all()
|
|
)
|
|
|
|
# Most traded securities
|
|
active_securities = (
|
|
self.session.query(
|
|
Security.ticker, func.count(Trade.id).label("trade_count")
|
|
)
|
|
.join(Trade)
|
|
.filter(Trade.filing_date >= week_ago)
|
|
.group_by(Security.id, Security.ticker)
|
|
.order_by(func.count(Trade.id).desc())
|
|
.limit(10)
|
|
.all()
|
|
)
|
|
|
|
# Get top suspicious patterns
|
|
repeat_offenders = self.detector.identify_repeat_offenders(
|
|
days_lookback=7, min_suspicious_trades=2, min_timing_score=40
|
|
)
|
|
|
|
return {
|
|
"period_start": week_ago,
|
|
"period_end": date.today(),
|
|
"most_active_officials": [
|
|
{"name": name, "trade_count": count} for name, count in active_officials
|
|
],
|
|
"most_traded_securities": [
|
|
{"ticker": ticker, "trade_count": count}
|
|
for ticker, count in active_securities
|
|
],
|
|
"repeat_offenders_count": len(repeat_offenders),
|
|
"repeat_offenders": repeat_offenders[:5], # Top 5
|
|
}
|
|
|
|
def format_as_text(self, report_data: Dict[str, Any], report_type: str) -> str:
|
|
"""
|
|
Format report data as plain text.
|
|
|
|
Args:
|
|
report_data: Report data dictionary
|
|
report_type: Type of report ('daily' or 'weekly')
|
|
|
|
Returns:
|
|
Formatted plain text report
|
|
"""
|
|
if report_type == "daily":
|
|
return self._format_daily_text(report_data)
|
|
elif report_type == "weekly":
|
|
return self._format_weekly_text(report_data)
|
|
else:
|
|
return str(report_data)
|
|
|
|
def _format_daily_text(self, data: Dict[str, Any]) -> str:
|
|
"""Format daily report as plain text."""
|
|
if data.get("lookback_days", 1) > 1:
|
|
trades_label = (
|
|
f" • Trades Filed (last {data['lookback_days']} days): {data['new_trades_count']}"
|
|
)
|
|
else:
|
|
trades_label = f" • New Trades Filed: {data['new_trades_count']}"
|
|
|
|
lines = [
|
|
"=" * 70,
|
|
f"POTE DAILY REPORT - {data['date']}",
|
|
"=" * 70,
|
|
"",
|
|
"📊 SUMMARY",
|
|
trades_label,
|
|
f" • Market Alerts: {data['market_alerts_count']}",
|
|
f" • Critical Alerts (≥7 severity): {data['critical_alerts_count']}",
|
|
f" • Suspicious Timing Trades: {data['suspicious_trades_count']}",
|
|
"",
|
|
]
|
|
|
|
if data["new_trades"]:
|
|
lines.append("📝 NEW TRADES")
|
|
for t in data["new_trades"][:10]: # Limit to 10
|
|
lines.append(
|
|
f" • {t['official']}: {t['side']} {t['ticker']} "
|
|
f"(${t['value_min']:,.0f} - ${t['value_max']:,.0f}) "
|
|
f"on {t['transaction_date']}"
|
|
)
|
|
if len(data["new_trades"]) > 10:
|
|
lines.append(f" ... and {len(data['new_trades']) - 10} more")
|
|
lines.append("")
|
|
|
|
if data["critical_alerts"]:
|
|
lines.append("🚨 CRITICAL MARKET ALERTS")
|
|
for a in data["critical_alerts"][:5]:
|
|
lines.append(
|
|
f" • {a['ticker']}: {a['type']} (severity {a['severity']}) "
|
|
f"at {a['timestamp'].strftime('%H:%M:%S')}"
|
|
)
|
|
lines.append("")
|
|
|
|
if data["suspicious_trades"]:
|
|
lines.append("⚠️ SUSPICIOUS TIMING DETECTED")
|
|
for st in data["suspicious_trades"][:5]:
|
|
lines.append(
|
|
f" • {st['official_name']}: {st['side']} {st['ticker']} "
|
|
f"(Timing Score: {st['timing_score']}/100, "
|
|
f"{st['prior_alerts_count']} prior alerts)"
|
|
)
|
|
lines.append("")
|
|
|
|
lines.extend(
|
|
[
|
|
"=" * 70,
|
|
"DISCLAIMER: This is for research purposes only. Not investment advice.",
|
|
"=" * 70,
|
|
]
|
|
)
|
|
|
|
return "\n".join(lines)
|
|
|
|
def _format_weekly_text(self, data: Dict[str, Any]) -> str:
|
|
"""Format weekly report as plain text."""
|
|
lines = [
|
|
"=" * 70,
|
|
f"POTE WEEKLY REPORT - {data['period_start']} to {data['period_end']}",
|
|
"=" * 70,
|
|
"",
|
|
"👥 MOST ACTIVE OFFICIALS",
|
|
]
|
|
|
|
for official in data["most_active_officials"]:
|
|
lines.append(f" • {official['name']}: {official['trade_count']} trades")
|
|
|
|
lines.extend(["", "📈 MOST TRADED SECURITIES"])
|
|
|
|
for security in data["most_traded_securities"]:
|
|
lines.append(f" • {security['ticker']}: {security['trade_count']} trades")
|
|
|
|
if data["repeat_offenders"]:
|
|
lines.extend(
|
|
["", f"⚠️ REPEAT OFFENDERS ({data['repeat_offenders_count']} total)"]
|
|
)
|
|
for offender in data["repeat_offenders"]:
|
|
lines.append(
|
|
f" • {offender['official_name']}: "
|
|
f"{offender['trades_with_timing_advantage']}/{offender['total_trades']} "
|
|
f"suspicious trades (avg score: {offender['average_timing_score']:.1f})"
|
|
)
|
|
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"=" * 70,
|
|
"DISCLAIMER: This is for research purposes only. Not investment advice.",
|
|
"=" * 70,
|
|
]
|
|
)
|
|
|
|
return "\n".join(lines)
|
|
|
|
def format_as_html(self, report_data: Dict[str, Any], report_type: str) -> str:
|
|
"""
|
|
Format report data as HTML.
|
|
|
|
Args:
|
|
report_data: Report data dictionary
|
|
report_type: Type of report ('daily' or 'weekly')
|
|
|
|
Returns:
|
|
Formatted HTML report
|
|
"""
|
|
if report_type == "daily":
|
|
return self._format_daily_html(report_data)
|
|
elif report_type == "weekly":
|
|
return self._format_weekly_html(report_data)
|
|
else:
|
|
return f"<pre>{report_data}</pre>"
|
|
|
|
def _format_daily_html(self, data: Dict[str, Any]) -> str:
|
|
"""Format daily report as HTML."""
|
|
if data.get("lookback_days", 1) > 1:
|
|
new_trades_label = f"Trades Filed (last {data['lookback_days']} days):"
|
|
else:
|
|
new_trades_label = "New Trades:"
|
|
|
|
html = f"""
|
|
<html>
|
|
<head>
|
|
<style>
|
|
body {{ font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }}
|
|
h1 {{ color: #2c3e50; border-bottom: 3px solid #3498db; padding-bottom: 10px; }}
|
|
h2 {{ color: #34495e; margin-top: 30px; }}
|
|
.summary {{ background: #ecf0f1; padding: 15px; border-radius: 5px; margin: 20px 0; }}
|
|
.stat {{ display: inline-block; margin-right: 20px; }}
|
|
.alert {{ background: #fff3cd; padding: 10px; margin: 5px 0; border-left: 4px solid #ffc107; }}
|
|
.critical {{ background: #f8d7da; border-left: 4px solid #dc3545; }}
|
|
.trade {{ background: #d1ecf1; padding: 10px; margin: 5px 0; border-left: 4px solid #17a2b8; }}
|
|
.disclaimer {{ background: #e9ecef; padding: 10px; margin-top: 30px; font-size: 0.9em; border-left: 4px solid #6c757d; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>POTE Daily Report - {data['date']}</h1>
|
|
|
|
<div class="summary">
|
|
<h2>📊 Summary</h2>
|
|
<div class="stat"><strong>{new_trades_label}</strong> {data['new_trades_count']}</div>
|
|
<div class="stat"><strong>Market Alerts:</strong> {data['market_alerts_count']}</div>
|
|
<div class="stat"><strong>Critical Alerts:</strong> {data['critical_alerts_count']}</div>
|
|
<div class="stat"><strong>Suspicious Trades:</strong> {data['suspicious_trades_count']}</div>
|
|
</div>
|
|
"""
|
|
|
|
if data["new_trades"]:
|
|
html += "<h2>📝 New Trades</h2>"
|
|
for t in data["new_trades"][:10]:
|
|
html += f"""
|
|
<div class="trade">
|
|
<strong>{t['official']}</strong>: {t['side']} {t['ticker']}
|
|
(${t['value_min']:,.0f} - ${t['value_max']:,.0f}) on {t['transaction_date']}
|
|
</div>
|
|
"""
|
|
|
|
if data["critical_alerts"]:
|
|
html += "<h2>🚨 Critical Market Alerts</h2>"
|
|
for a in data["critical_alerts"][:5]:
|
|
html += f"""
|
|
<div class="alert critical">
|
|
<strong>{a['ticker']}</strong>: {a['type']} (severity {a['severity']})
|
|
at {a['timestamp'].strftime('%H:%M:%S')}
|
|
</div>
|
|
"""
|
|
|
|
if data["suspicious_trades"]:
|
|
html += "<h2>⚠️ Suspicious Timing Detected</h2>"
|
|
for st in data["suspicious_trades"][:5]:
|
|
html += f"""
|
|
<div class="alert">
|
|
<strong>{st['official_name']}</strong>: {st['side']} {st['ticker']}<br>
|
|
Timing Score: {st['timing_score']}/100 ({st['prior_alerts_count']} prior alerts)
|
|
</div>
|
|
"""
|
|
|
|
html += """
|
|
<div class="disclaimer">
|
|
<strong>DISCLAIMER:</strong> This is for research purposes only. Not investment advice.
|
|
</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
return html
|
|
|
|
def _format_weekly_html(self, data: Dict[str, Any]) -> str:
|
|
"""Format weekly report as HTML."""
|
|
html = f"""
|
|
<html>
|
|
<head>
|
|
<style>
|
|
body {{ font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }}
|
|
h1 {{ color: #2c3e50; border-bottom: 3px solid #3498db; padding-bottom: 10px; }}
|
|
h2 {{ color: #34495e; margin-top: 30px; }}
|
|
table {{ width: 100%; border-collapse: collapse; margin: 20px 0; }}
|
|
th, td {{ padding: 10px; text-align: left; border-bottom: 1px solid #ddd; }}
|
|
th {{ background: #3498db; color: white; }}
|
|
.disclaimer {{ background: #e9ecef; padding: 10px; margin-top: 30px; font-size: 0.9em; border-left: 4px solid #6c757d; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>POTE Weekly Report</h1>
|
|
<p><strong>Period:</strong> {data['period_start']} to {data['period_end']}</p>
|
|
|
|
<h2>👥 Most Active Officials</h2>
|
|
<table>
|
|
<tr><th>Official</th><th>Trade Count</th></tr>
|
|
"""
|
|
|
|
for official in data["most_active_officials"]:
|
|
html += f"<tr><td>{official['name']}</td><td>{official['trade_count']}</td></tr>"
|
|
|
|
html += """
|
|
</table>
|
|
|
|
<h2>📈 Most Traded Securities</h2>
|
|
<table>
|
|
<tr><th>Ticker</th><th>Trade Count</th></tr>
|
|
"""
|
|
|
|
for security in data["most_traded_securities"]:
|
|
html += f"<tr><td>{security['ticker']}</td><td>{security['trade_count']}</td></tr>"
|
|
|
|
html += "</table>"
|
|
|
|
if data["repeat_offenders"]:
|
|
html += f"""
|
|
<h2>⚠️ Repeat Offenders ({data['repeat_offenders_count']} total)</h2>
|
|
<table>
|
|
<tr><th>Official</th><th>Suspicious Trades</th><th>Total Trades</th><th>Avg Score</th></tr>
|
|
"""
|
|
for offender in data["repeat_offenders"]:
|
|
html += f"""
|
|
<tr>
|
|
<td>{offender['official_name']}</td>
|
|
<td>{offender['trades_with_timing_advantage']}</td>
|
|
<td>{offender['total_trades']}</td>
|
|
<td>{offender['average_timing_score']:.1f}</td>
|
|
</tr>
|
|
"""
|
|
html += "</table>"
|
|
|
|
html += """
|
|
<div class="disclaimer">
|
|
<strong>DISCLAIMER:</strong> This is for research purposes only. Not investment advice.
|
|
</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
return html
|
|
|