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
336 lines
11 KiB
Python
336 lines
11 KiB
Python
"""Integration tests for analytics with real-ish data."""
|
|
|
|
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from pote.analytics.benchmarks import BenchmarkComparison
|
|
from pote.analytics.metrics import PerformanceMetrics
|
|
from pote.analytics.returns import ReturnCalculator
|
|
from pote.db.models import Official, Price, Security, Trade
|
|
|
|
|
|
@pytest.fixture
|
|
def full_test_data(test_db_session):
|
|
"""Create complete test dataset with prices."""
|
|
session = test_db_session
|
|
# Create officials
|
|
pelosi = Official(
|
|
name="Nancy Pelosi",
|
|
chamber="House",
|
|
party="Democrat",
|
|
state="CA",
|
|
)
|
|
tuberville = Official(
|
|
name="Tommy Tuberville",
|
|
chamber="Senate",
|
|
party="Republican",
|
|
state="AL",
|
|
)
|
|
session.add_all([pelosi, tuberville])
|
|
session.flush()
|
|
|
|
# Create securities
|
|
nvda = Security(ticker="NVDA", name="NVIDIA Corporation", sector="Technology")
|
|
spy = Security(ticker="SPY", name="SPDR S&P 500 ETF", sector="Financial")
|
|
session.add_all([nvda, spy])
|
|
session.flush()
|
|
|
|
# Create price data for NVDA (upward trend)
|
|
base_date = date(2024, 1, 1)
|
|
nvda_base_price = Decimal("495.00")
|
|
|
|
for i in range(120):
|
|
current_date = base_date + timedelta(days=i)
|
|
# Simulate upward trend: +0.5% per day on average
|
|
price_change = Decimal(i) * Decimal("2.50") # ~50% gain over 120 days
|
|
current_price = nvda_base_price + price_change
|
|
|
|
price = Price(
|
|
security_id=nvda.id,
|
|
date=current_date,
|
|
open=current_price - Decimal("5.00"),
|
|
high=current_price + Decimal("5.00"),
|
|
low=current_price - Decimal("8.00"),
|
|
close=current_price,
|
|
volume=50000000,
|
|
)
|
|
session.add(price)
|
|
|
|
# Create price data for SPY (slower upward trend - ~10% over 120 days)
|
|
spy_base_price = Decimal("450.00")
|
|
|
|
for i in range(120):
|
|
current_date = base_date + timedelta(days=i)
|
|
price_change = Decimal(i) * Decimal("0.35")
|
|
current_price = spy_base_price + price_change
|
|
|
|
price = Price(
|
|
security_id=spy.id,
|
|
date=current_date,
|
|
open=current_price - Decimal("2.00"),
|
|
high=current_price + Decimal("2.00"),
|
|
low=current_price - Decimal("3.00"),
|
|
close=current_price,
|
|
volume=100000000,
|
|
)
|
|
session.add(price)
|
|
|
|
# Create trades
|
|
# Pelosi buys NVDA early (should show good returns)
|
|
trade1 = Trade(
|
|
official_id=pelosi.id,
|
|
security_id=nvda.id,
|
|
source="test",
|
|
transaction_date=date(2024, 1, 15), # Buy at ~495
|
|
filing_date=date(2024, 2, 1),
|
|
side="buy",
|
|
value_min=Decimal("15001"),
|
|
value_max=Decimal("50000"),
|
|
)
|
|
|
|
# Tuberville buys NVDA later (still good but less alpha)
|
|
trade2 = Trade(
|
|
official_id=tuberville.id,
|
|
security_id=nvda.id,
|
|
source="test",
|
|
transaction_date=date(2024, 2, 1), # Buy at ~540
|
|
filing_date=date(2024, 2, 15),
|
|
side="buy",
|
|
value_min=Decimal("50001"),
|
|
value_max=Decimal("100000"),
|
|
)
|
|
|
|
session.add_all([trade1, trade2])
|
|
session.commit()
|
|
|
|
return {
|
|
"officials": [pelosi, tuberville],
|
|
"securities": [nvda, spy],
|
|
"trades": [trade1, trade2],
|
|
}
|
|
|
|
|
|
def test_return_calculation_with_real_data(test_db_session, full_test_data):
|
|
session = test_db_session
|
|
"""Test return calculation with realistic price data."""
|
|
calculator = ReturnCalculator(session)
|
|
|
|
# Get Pelosi's NVDA trade
|
|
trade = full_test_data["trades"][0]
|
|
|
|
# Calculate 90-day return
|
|
result = calculator.calculate_trade_return(trade, window_days=90)
|
|
|
|
assert result is not None, "Should calculate return with available data"
|
|
assert result["ticker"] == "NVDA"
|
|
assert result["window_days"] == 90
|
|
assert result["return_pct"] > 0, "NVDA should have positive return"
|
|
|
|
# Entry around day 15, exit around day 105
|
|
# Expected return: (720 - 532.5) / 532.5 = ~35%
|
|
assert (
|
|
30 < float(result["return_pct"]) < 50
|
|
), f"Expected ~35% return, got {result['return_pct']}"
|
|
|
|
print(f"\n✅ NVDA 90-day return: {result['return_pct']:.2f}%")
|
|
print(f" Entry: ${result['entry_price']} on {result['transaction_date']}")
|
|
print(f" Exit: ${result['exit_price']} on {result['exit_date']}")
|
|
|
|
|
|
def test_benchmark_comparison_with_real_data(test_db_session, full_test_data):
|
|
session = test_db_session
|
|
"""Test benchmark comparison with SPY."""
|
|
benchmark = BenchmarkComparison(session)
|
|
|
|
# Get Pelosi's trade
|
|
trade = full_test_data["trades"][0]
|
|
|
|
# Compare to SPY
|
|
result = benchmark.compare_trade_to_benchmark(trade, window_days=90, benchmark="SPY")
|
|
|
|
assert result is not None
|
|
assert result["ticker"] == "NVDA"
|
|
assert result["benchmark"] == "SPY"
|
|
|
|
# NVDA should beat SPY significantly
|
|
assert result["beat_market"] is True
|
|
assert float(result["abnormal_return"]) > 10, "NVDA should have strong alpha vs SPY"
|
|
|
|
print("\n✅ Benchmark Comparison:")
|
|
print(f" NVDA Return: {result['trade_return']:.2f}%")
|
|
print(f" SPY Return: {result['benchmark_return']:.2f}%")
|
|
print(f" Alpha: {result['abnormal_return']:+.2f}%")
|
|
|
|
|
|
def test_official_performance_summary(test_db_session, full_test_data):
|
|
session = test_db_session
|
|
"""Test official performance aggregation."""
|
|
metrics = PerformanceMetrics(session)
|
|
|
|
pelosi = full_test_data["officials"][0]
|
|
|
|
# Get performance summary
|
|
perf = metrics.official_performance(pelosi.id, window_days=90)
|
|
|
|
assert perf["name"] == "Nancy Pelosi"
|
|
assert perf["total_trades"] >= 1
|
|
|
|
if perf.get("trades_analyzed", 0) > 0:
|
|
assert "avg_return" in perf
|
|
assert "avg_alpha" in perf
|
|
assert "win_rate" in perf
|
|
assert perf["win_rate"] >= 0 and perf["win_rate"] <= 1
|
|
|
|
print(f"\n✅ {perf['name']} Performance:")
|
|
print(f" Total Trades: {perf['total_trades']}")
|
|
print(f" Average Return: {perf['avg_return']:.2f}%")
|
|
print(f" Alpha: {perf['avg_alpha']:+.2f}%")
|
|
print(f" Win Rate: {perf['win_rate']:.1%}")
|
|
|
|
|
|
def test_multiple_windows(test_db_session, full_test_data):
|
|
session = test_db_session
|
|
"""Test calculating multiple time windows."""
|
|
calculator = ReturnCalculator(session)
|
|
|
|
trade = full_test_data["trades"][0]
|
|
|
|
# Calculate for 30, 60, 90 days
|
|
results = calculator.calculate_multiple_windows(trade, windows=[30, 60, 90])
|
|
|
|
assert len(results) == 3, "Should calculate all three windows"
|
|
|
|
# Returns should generally increase with longer windows (given upward trend)
|
|
if 30 in results and 90 in results:
|
|
print("\n✅ Multiple Windows:")
|
|
for window in [30, 60, 90]:
|
|
if window in results:
|
|
print(f" {window:3d} days: {results[window]['return_pct']:+7.2f}%")
|
|
|
|
|
|
def test_top_performers(test_db_session, full_test_data):
|
|
session = test_db_session
|
|
"""Test top performer ranking."""
|
|
metrics = PerformanceMetrics(session)
|
|
|
|
top = metrics.top_performers(window_days=90, limit=5)
|
|
|
|
assert isinstance(top, list)
|
|
assert len(top) > 0
|
|
|
|
print("\n✅ Top Performers:")
|
|
for i, perf in enumerate(top, 1):
|
|
if perf.get("trades_analyzed", 0) > 0:
|
|
print(f" {i}. {perf['name']:20s} | Alpha: {perf['avg_alpha']:+6.2f}%")
|
|
|
|
|
|
def test_system_statistics(test_db_session, full_test_data):
|
|
session = test_db_session
|
|
"""Test system-wide statistics."""
|
|
metrics = PerformanceMetrics(session)
|
|
|
|
stats = metrics.summary_statistics(window_days=90)
|
|
|
|
assert stats["total_officials"] >= 2
|
|
assert stats["total_trades"] >= 2
|
|
assert stats["total_securities"] >= 2
|
|
|
|
print("\n✅ System Statistics:")
|
|
print(f" Officials: {stats['total_officials']}")
|
|
print(f" Trades: {stats['total_trades']}")
|
|
print(f" Securities: {stats['total_securities']}")
|
|
|
|
if stats.get("avg_alpha") is not None:
|
|
print(f" Avg Alpha: {stats['avg_alpha']:+.2f}%")
|
|
print(f" Beat Market: {stats['beat_market_rate']:.1%}")
|
|
|
|
|
|
def test_disclosure_timing(test_db_session, full_test_data):
|
|
session = test_db_session
|
|
"""Test disclosure lag analysis."""
|
|
metrics = PerformanceMetrics(session)
|
|
|
|
timing = metrics.timing_analysis()
|
|
|
|
assert "avg_disclosure_lag_days" in timing
|
|
assert timing["avg_disclosure_lag_days"] > 0
|
|
|
|
print("\n✅ Disclosure Timing:")
|
|
print(f" Average Lag: {timing['avg_disclosure_lag_days']:.1f} days")
|
|
print(f" Median Lag: {timing['median_disclosure_lag_days']} days")
|
|
|
|
|
|
def test_sector_analysis(test_db_session, full_test_data):
|
|
session = test_db_session
|
|
"""Test sector-level analysis."""
|
|
metrics = PerformanceMetrics(session)
|
|
|
|
sectors = metrics.sector_analysis(window_days=90)
|
|
|
|
assert isinstance(sectors, list)
|
|
|
|
if sectors:
|
|
print("\n✅ Sector Analysis:")
|
|
for s in sectors:
|
|
print(
|
|
f" {s['sector']:15s} | {s['trade_count']} trades | Alpha: {s['avg_alpha']:+6.2f}%"
|
|
)
|
|
|
|
|
|
def test_edge_case_missing_exit_price(test_db_session, full_test_data):
|
|
session = test_db_session
|
|
"""Test handling of trade with no exit price available."""
|
|
calculator = ReturnCalculator(session)
|
|
|
|
nvda = session.query(Security).filter_by(ticker="NVDA").first()
|
|
pelosi = session.query(Official).filter_by(name="Nancy Pelosi").first()
|
|
|
|
# Create trade with transaction date far in future (no exit price)
|
|
future_trade = Trade(
|
|
official_id=pelosi.id,
|
|
security_id=nvda.id,
|
|
source="test",
|
|
transaction_date=date(2024, 12, 1), # No prices available this far out
|
|
side="buy",
|
|
value_min=Decimal("10000"),
|
|
value_max=Decimal("50000"),
|
|
)
|
|
session.add(future_trade)
|
|
session.commit()
|
|
|
|
result = calculator.calculate_trade_return(future_trade, window_days=90)
|
|
|
|
assert result is None, "Should return None when price data unavailable"
|
|
print("\n✅ Correctly handles missing price data")
|
|
|
|
|
|
def test_sell_trade_logic(test_db_session, full_test_data):
|
|
session = test_db_session
|
|
"""Test that sell trades have inverted return logic."""
|
|
calculator = ReturnCalculator(session)
|
|
|
|
nvda = session.query(Security).filter_by(ticker="NVDA").first()
|
|
pelosi = session.query(Official).filter_by(name="Nancy Pelosi").first()
|
|
|
|
# Create sell trade during uptrend (should show negative return)
|
|
sell_trade = Trade(
|
|
official_id=pelosi.id,
|
|
security_id=nvda.id,
|
|
source="test",
|
|
transaction_date=date(2024, 1, 15),
|
|
side="sell",
|
|
value_min=Decimal("10000"),
|
|
value_max=Decimal("50000"),
|
|
)
|
|
session.add(sell_trade)
|
|
session.commit()
|
|
|
|
result = calculator.calculate_trade_return(sell_trade, window_days=90)
|
|
|
|
if result:
|
|
# Selling during uptrend = negative return
|
|
assert result["return_pct"] < 0, "Sell during uptrend should show negative return"
|
|
print(f"\n✅ Sell trade return correctly inverted: {result['return_pct']:.2f}%")
|