ilia
5df21d82f4
- ruff: fix all 328 errors (autofix + manual); move config to [tool.ruff.lint] - mypy: fix all errors (annotations, nullable-column guards, stale kwargs in weekly report caller) - black: reformat src/tests so black --check passes - CI: remove `|| true` from ruff/black/mypy/pytest steps in both .gitea and .github workflows; install project deps and add mypy step in gitea python-ci so gates actually run - docs: move 14 root status/guide markdown files into docs/, update links
455 lines
14 KiB
Python
455 lines
14 KiB
Python
"""Tests for disclosure correlation module."""
|
|
|
|
from datetime import UTC, date, datetime
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from pote.db.models import MarketAlert, Official, Security, Trade
|
|
from pote.monitoring.disclosure_correlator import DisclosureCorrelator
|
|
|
|
|
|
@pytest.fixture
|
|
def trade_with_alerts(test_db_session):
|
|
"""Create a trade with prior market alerts."""
|
|
session = test_db_session
|
|
|
|
# Create official and security
|
|
pelosi = Official(name="Nancy Pelosi", chamber="House", party="Democrat", state="CA")
|
|
nvda = Security(ticker="NVDA", name="NVIDIA Corporation", sector="Technology")
|
|
session.add_all([pelosi, nvda])
|
|
session.flush()
|
|
|
|
# Create trade on Jan 15
|
|
trade = Trade(
|
|
official_id=pelosi.id,
|
|
security_id=nvda.id,
|
|
source="test",
|
|
transaction_date=date(2024, 1, 15),
|
|
filing_date=date(2024, 2, 1),
|
|
side="buy",
|
|
value_min=Decimal("15001"),
|
|
value_max=Decimal("50000"),
|
|
)
|
|
session.add(trade)
|
|
session.flush()
|
|
|
|
# Create alerts BEFORE trade (suspicious)
|
|
alerts = [
|
|
MarketAlert(
|
|
ticker="NVDA",
|
|
alert_type="unusual_volume",
|
|
timestamp=datetime(2024, 1, 10, 10, 30, tzinfo=UTC), # 5 days before
|
|
details={"multiplier": 3.5},
|
|
price=Decimal("490.00"),
|
|
volume=100000000,
|
|
change_pct=Decimal("2.0"),
|
|
severity=8,
|
|
),
|
|
MarketAlert(
|
|
ticker="NVDA",
|
|
alert_type="price_spike",
|
|
timestamp=datetime(2024, 1, 12, 14, 15, tzinfo=UTC), # 3 days before
|
|
details={"change_pct": 5.5},
|
|
price=Decimal("505.00"),
|
|
volume=85000000,
|
|
change_pct=Decimal("5.5"),
|
|
severity=7,
|
|
),
|
|
MarketAlert(
|
|
ticker="NVDA",
|
|
alert_type="high_volatility",
|
|
timestamp=datetime(2024, 1, 14, 16, 20, tzinfo=UTC), # 1 day before
|
|
details={"multiplier": 2.5},
|
|
price=Decimal("510.00"),
|
|
volume=90000000,
|
|
change_pct=Decimal("1.5"),
|
|
severity=6,
|
|
),
|
|
]
|
|
session.add_all(alerts)
|
|
session.commit()
|
|
|
|
return {
|
|
"trade": trade,
|
|
"official": pelosi,
|
|
"security": nvda,
|
|
"alerts": alerts,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def trade_without_alerts(test_db_session):
|
|
"""Create a trade without prior alerts (clean)."""
|
|
session = test_db_session
|
|
|
|
official = Official(name="John Smith", chamber="House", party="Republican", state="TX")
|
|
security = Security(ticker="MSFT", name="Microsoft Corporation", sector="Technology")
|
|
session.add_all([official, security])
|
|
session.flush()
|
|
|
|
trade = Trade(
|
|
official_id=official.id,
|
|
security_id=security.id,
|
|
source="test",
|
|
transaction_date=date(2024, 2, 1),
|
|
side="buy",
|
|
value_min=Decimal("10000"),
|
|
value_max=Decimal("50000"),
|
|
)
|
|
session.add(trade)
|
|
session.commit()
|
|
|
|
return {
|
|
"trade": trade,
|
|
"official": official,
|
|
"security": security,
|
|
}
|
|
|
|
|
|
def test_get_alerts_before_trade(test_db_session, trade_with_alerts):
|
|
"""Test retrieving alerts before a trade."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
trade = trade_with_alerts["trade"]
|
|
|
|
# Get alerts before trade
|
|
prior_alerts = correlator.get_alerts_before_trade(trade, lookback_days=30)
|
|
|
|
assert len(prior_alerts) == 3
|
|
assert all(alert.ticker == "NVDA" for alert in prior_alerts)
|
|
assert all(alert.timestamp.date() < trade.transaction_date for alert in prior_alerts)
|
|
|
|
|
|
def test_get_alerts_before_trade_no_alerts(test_db_session, trade_without_alerts):
|
|
"""Test retrieving alerts when none exist."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
trade = trade_without_alerts["trade"]
|
|
|
|
prior_alerts = correlator.get_alerts_before_trade(trade, lookback_days=30)
|
|
|
|
assert len(prior_alerts) == 0
|
|
|
|
|
|
def test_calculate_timing_score_high_suspicion(test_db_session, trade_with_alerts):
|
|
"""Test timing score calculation for suspicious trade."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
trade = trade_with_alerts["trade"]
|
|
alerts = trade_with_alerts["alerts"]
|
|
|
|
timing_analysis = correlator.calculate_timing_score(trade, alerts)
|
|
|
|
assert timing_analysis["timing_score"] > 60, "Should be suspicious with 3 alerts"
|
|
assert timing_analysis["suspicious"] is True
|
|
assert timing_analysis["alert_count"] == 3
|
|
assert timing_analysis["recent_alert_count"] > 0
|
|
assert timing_analysis["high_severity_count"] >= 2 # 2 alerts with severity 7+
|
|
assert "reason" in timing_analysis
|
|
|
|
|
|
def test_calculate_timing_score_no_alerts(test_db_session):
|
|
"""Test timing score with no prior alerts."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
# Create minimal trade
|
|
official = Official(name="Test", chamber="House", party="Democrat", state="CA")
|
|
security = Security(ticker="TEST", name="Test Corp")
|
|
session.add_all([official, security])
|
|
session.flush()
|
|
|
|
trade = Trade(
|
|
official_id=official.id,
|
|
security_id=security.id,
|
|
source="test",
|
|
transaction_date=date(2024, 1, 1),
|
|
side="buy",
|
|
value_min=Decimal("10000"),
|
|
)
|
|
session.add(trade)
|
|
session.commit()
|
|
|
|
timing_analysis = correlator.calculate_timing_score(trade, [])
|
|
|
|
assert timing_analysis["timing_score"] == 0
|
|
assert timing_analysis["suspicious"] is False
|
|
assert timing_analysis["alert_count"] == 0
|
|
|
|
|
|
def test_calculate_timing_score_factors(test_db_session):
|
|
"""Test that timing score considers all factors correctly."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
# Create trade
|
|
official = Official(name="Test", chamber="House", party="Democrat", state="CA")
|
|
security = Security(ticker="TEST", name="Test Corp")
|
|
session.add_all([official, security])
|
|
session.flush()
|
|
|
|
trade_date = date(2024, 1, 15)
|
|
trade = Trade(
|
|
official_id=official.id,
|
|
security_id=security.id,
|
|
source="test",
|
|
transaction_date=trade_date,
|
|
side="buy",
|
|
value_min=Decimal("10000"),
|
|
)
|
|
session.add(trade)
|
|
session.flush()
|
|
|
|
# Test with low severity alerts (should have lower score)
|
|
low_sev_alerts = [
|
|
MarketAlert(
|
|
ticker="TEST",
|
|
alert_type="test",
|
|
timestamp=datetime(2024, 1, 10, 12, 0, tzinfo=UTC),
|
|
severity=3,
|
|
),
|
|
MarketAlert(
|
|
ticker="TEST",
|
|
alert_type="test",
|
|
timestamp=datetime(2024, 1, 11, 12, 0, tzinfo=UTC),
|
|
severity=4,
|
|
),
|
|
]
|
|
session.add_all(low_sev_alerts)
|
|
session.commit()
|
|
|
|
low_score = correlator.calculate_timing_score(trade, low_sev_alerts)
|
|
|
|
# Test with high severity alerts (should have higher score)
|
|
high_sev_alerts = [
|
|
MarketAlert(
|
|
ticker="TEST",
|
|
alert_type="test",
|
|
timestamp=datetime(2024, 1, 13, 12, 0, tzinfo=UTC), # Recent
|
|
severity=9,
|
|
),
|
|
MarketAlert(
|
|
ticker="TEST",
|
|
alert_type="test",
|
|
timestamp=datetime(2024, 1, 14, 12, 0, tzinfo=UTC), # Very recent
|
|
severity=8,
|
|
),
|
|
]
|
|
session.add_all(high_sev_alerts)
|
|
session.commit()
|
|
|
|
high_score = correlator.calculate_timing_score(trade, high_sev_alerts)
|
|
|
|
# High severity + recent should score higher
|
|
assert high_score["timing_score"] > low_score["timing_score"]
|
|
assert high_score["recent_alert_count"] > 0
|
|
assert high_score["high_severity_count"] > 0
|
|
|
|
|
|
def test_analyze_trade_full(test_db_session, trade_with_alerts):
|
|
"""Test complete trade analysis."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
trade = trade_with_alerts["trade"]
|
|
|
|
analysis = correlator.analyze_trade(trade)
|
|
|
|
# Check all required fields
|
|
assert analysis["trade_id"] == trade.id
|
|
assert analysis["official_name"] == "Nancy Pelosi"
|
|
assert analysis["ticker"] == "NVDA"
|
|
assert analysis["side"] == "buy"
|
|
assert analysis["transaction_date"] == "2024-01-15"
|
|
assert analysis["timing_score"] > 0
|
|
assert "prior_alerts" in analysis
|
|
assert len(analysis["prior_alerts"]) == 3
|
|
|
|
# Check alert details
|
|
for alert_detail in analysis["prior_alerts"]:
|
|
assert "timestamp" in alert_detail
|
|
assert "alert_type" in alert_detail
|
|
assert "severity" in alert_detail
|
|
assert "days_before_trade" in alert_detail
|
|
assert alert_detail["days_before_trade"] >= 0
|
|
|
|
|
|
def test_analyze_recent_disclosures(test_db_session, trade_with_alerts, trade_without_alerts):
|
|
"""Test batch analysis of recent disclosures."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
# Both trades were created "recently" (in fixture setup)
|
|
suspicious_trades = correlator.analyze_recent_disclosures(
|
|
days=365, min_timing_score=50 # Wide window to catch test data
|
|
)
|
|
|
|
# Should find at least the suspicious trade
|
|
assert len(suspicious_trades) >= 1
|
|
|
|
# Check sorting (highest score first)
|
|
if len(suspicious_trades) > 1:
|
|
for i in range(len(suspicious_trades) - 1):
|
|
assert suspicious_trades[i]["timing_score"] >= suspicious_trades[i + 1]["timing_score"]
|
|
|
|
|
|
def test_get_official_timing_pattern(test_db_session, trade_with_alerts):
|
|
"""Test official timing pattern analysis."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
official = trade_with_alerts["official"]
|
|
|
|
# Use wide lookback to catch test data (trade is 2024-01-15)
|
|
pattern = correlator.get_official_timing_pattern(official.id, lookback_days=3650)
|
|
|
|
assert pattern["official_id"] == official.id
|
|
assert pattern["trade_count"] >= 1
|
|
assert pattern["trades_with_prior_alerts"] >= 1
|
|
assert pattern["suspicious_trade_count"] >= 0
|
|
assert "pattern" in pattern
|
|
assert "analyses" in pattern
|
|
|
|
|
|
def test_get_official_timing_pattern_no_trades(test_db_session):
|
|
"""Test official with no trades."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
official = Official(name="No Trades", chamber="House", party="Democrat", state="CA")
|
|
session.add(official)
|
|
session.commit()
|
|
|
|
pattern = correlator.get_official_timing_pattern(official.id)
|
|
|
|
assert pattern["trade_count"] == 0
|
|
assert "No trades" in pattern["pattern"]
|
|
|
|
|
|
def test_get_ticker_timing_analysis(test_db_session, trade_with_alerts):
|
|
"""Test ticker timing analysis."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
# Use wide lookback to catch test data
|
|
analysis = correlator.get_ticker_timing_analysis("NVDA", lookback_days=3650)
|
|
|
|
assert analysis["ticker"] == "NVDA"
|
|
assert analysis["trade_count"] >= 1
|
|
assert analysis["trades_with_alerts"] >= 1
|
|
assert "avg_timing_score" in analysis
|
|
assert "analyses" in analysis
|
|
|
|
|
|
def test_get_ticker_timing_analysis_no_trades(test_db_session):
|
|
"""Test ticker with no trades."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
analysis = correlator.get_ticker_timing_analysis("ZZZZ")
|
|
|
|
assert analysis["ticker"] == "ZZZZ"
|
|
assert analysis["trade_count"] == 0
|
|
assert "No trades" in analysis["pattern"]
|
|
|
|
|
|
def test_alerts_outside_lookback_window(test_db_session):
|
|
"""Test that alerts outside lookback window are excluded."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
# Create trade and alerts
|
|
official = Official(name="Test", chamber="House", party="Democrat", state="CA")
|
|
security = Security(ticker="TEST", name="Test Corp")
|
|
session.add_all([official, security])
|
|
session.flush()
|
|
|
|
trade_date = date(2024, 1, 15)
|
|
trade = Trade(
|
|
official_id=official.id,
|
|
security_id=security.id,
|
|
source="test",
|
|
transaction_date=trade_date,
|
|
side="buy",
|
|
value_min=Decimal("10000"),
|
|
)
|
|
session.add(trade)
|
|
session.flush()
|
|
|
|
# Alert 2 days before (within window)
|
|
recent_alert = MarketAlert(
|
|
ticker="TEST",
|
|
alert_type="test",
|
|
timestamp=datetime(2024, 1, 13, 12, 0, tzinfo=UTC),
|
|
severity=7,
|
|
)
|
|
|
|
# Alert 40 days before (outside 30-day window)
|
|
old_alert = MarketAlert(
|
|
ticker="TEST",
|
|
alert_type="test",
|
|
timestamp=datetime(2023, 12, 6, 12, 0, tzinfo=UTC),
|
|
severity=8,
|
|
)
|
|
|
|
session.add_all([recent_alert, old_alert])
|
|
session.commit()
|
|
|
|
# Should only get recent alert
|
|
alerts = correlator.get_alerts_before_trade(trade, lookback_days=30)
|
|
|
|
assert len(alerts) == 1
|
|
assert alerts[0].timestamp.date() == date(2024, 1, 13)
|
|
|
|
|
|
def test_different_ticker_alerts_excluded(test_db_session):
|
|
"""Test that alerts for different tickers are excluded."""
|
|
session = test_db_session
|
|
correlator = DisclosureCorrelator(session)
|
|
|
|
# Create trade for NVDA
|
|
official = Official(name="Test", chamber="House", party="Democrat", state="CA")
|
|
nvda = Security(ticker="NVDA", name="NVIDIA")
|
|
msft = Security(ticker="MSFT", name="Microsoft")
|
|
session.add_all([official, nvda, msft])
|
|
session.flush()
|
|
|
|
trade = Trade(
|
|
official_id=official.id,
|
|
security_id=nvda.id,
|
|
source="test",
|
|
transaction_date=date(2024, 1, 15),
|
|
side="buy",
|
|
value_min=Decimal("10000"),
|
|
)
|
|
session.add(trade)
|
|
session.flush()
|
|
|
|
# Create alerts for both tickers
|
|
nvda_alert = MarketAlert(
|
|
ticker="NVDA",
|
|
alert_type="test",
|
|
timestamp=datetime(2024, 1, 10, 12, 0, tzinfo=UTC),
|
|
severity=7,
|
|
)
|
|
|
|
msft_alert = MarketAlert(
|
|
ticker="MSFT",
|
|
alert_type="test",
|
|
timestamp=datetime(2024, 1, 10, 12, 0, tzinfo=UTC),
|
|
severity=8,
|
|
)
|
|
|
|
session.add_all([nvda_alert, msft_alert])
|
|
session.commit()
|
|
|
|
# Should only get NVDA alert
|
|
alerts = correlator.get_alerts_before_trade(trade, lookback_days=30)
|
|
|
|
assert len(alerts) == 1
|
|
assert alerts[0].ticker == "NVDA"
|