diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 500065d..1b86f40 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -46,10 +46,16 @@ jobs: python3 -m pip install --upgrade pip --break-system-packages if [ -f requirements.txt ]; then pip install -r requirements.txt --break-system-packages; fi if [ -f requirements-dev.txt ]; then pip install -r requirements-dev.txt --break-system-packages; fi - pip install bandit pip-audit ruff --break-system-packages + # Install the project + dev tools so lint/type/test gates run for real + pip install -e ".[dev]" --break-system-packages + pip install bandit pip-audit --break-system-packages + # Lint, type-check, and tests are hard gates ā€” no `|| true`. - name: Ruff lint - run: ruff check . || true + run: ruff check src tests + + - name: Mypy + run: mypy src - name: Bandit (advisory) run: bandit -r . -q || true @@ -58,13 +64,7 @@ jobs: run: pip-audit -r requirements.txt 2>/dev/null || pip-audit 2>/dev/null || true - name: Pytest - run: | - if [ -d tests ] || ls test_*.py *_test.py 2>/dev/null; then - pip install pytest --break-system-packages - pytest -q || true - else - echo "No tests found ā€” skip" - fi + run: pytest -q secret-scan: needs: skip-ci-check diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 76a8590..132269b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,14 +34,15 @@ jobs: .venv/bin/pip install --upgrade pip .venv/bin/pip install -e ".[dev]" + # Linters are hard gates ā€” no `|| true`. - name: Run linters run: | echo "Running ruff..." - .venv/bin/ruff check src/ tests/ || true + .venv/bin/ruff check src/ tests/ echo "Running black check..." - .venv/bin/black --check src/ tests/ || true + .venv/bin/black --check src/ tests/ echo "Running mypy..." - .venv/bin/mypy src/ --install-types --non-interactive || true + .venv/bin/mypy src/ --install-types --non-interactive - name: Run tests with coverage env: diff --git a/README.md b/README.md index 1625651..808c35a 100644 --- a/README.md +++ b/README.md @@ -24,11 +24,11 @@ POTE tracks stock trading activity of government officials (starting with U.S. C ## Quick start -**šŸš€ Already deployed?** See **[QUICKSTART.md](QUICKSTART.md)** for full usage guide! +**šŸš€ Already deployed?** See **[QUICKSTART.md](docs/QUICKSTART.md)** for full usage guide! -**šŸ“¦ Deploying?** See **[PROXMOX_QUICKSTART.md](PROXMOX_QUICKSTART.md)** for Proxmox LXC deployment (recommended). +**šŸ“¦ Deploying?** See **[PROXMOX_QUICKSTART.md](docs/PROXMOX_QUICKSTART.md)** for Proxmox LXC deployment (recommended). -**šŸ“§ Want automated reports?** See **[AUTOMATION_QUICKSTART.md](AUTOMATION_QUICKSTART.md)** for email reporting setup! +**šŸ“§ Want automated reports?** See **[AUTOMATION_QUICKSTART.md](docs/AUTOMATION_QUICKSTART.md)** for email reporting setup! **šŸ  Homelab deploy (LXC 236)?** See **[docs/HANDOFF-2026-05-27.md](docs/HANDOFF-2026-05-27.md)** for ops handoff and next steps. @@ -81,14 +81,14 @@ docker-compose up -d **Getting Started**: - [`README.md`](README.md) ā€“ This file -- [`QUICKSTART.md`](QUICKSTART.md) ā€“ ā­ **How to use your deployed POTE instance** -- [`STATUS.md`](STATUS.md) ā€“ Current project status -- [`FREE_TESTING_QUICKSTART.md`](FREE_TESTING_QUICKSTART.md) ā€“ Test for $0 -- [`OFFLINE_DEMO.md`](OFFLINE_DEMO.md) ā€“ Works without internet! +- [`QUICKSTART.md`](docs/QUICKSTART.md) ā€“ ā­ **How to use your deployed POTE instance** +- [`STATUS.md`](docs/STATUS.md) ā€“ Current project status +- [`FREE_TESTING_QUICKSTART.md`](docs/FREE_TESTING_QUICKSTART.md) ā€“ Test for $0 +- [`OFFLINE_DEMO.md`](docs/OFFLINE_DEMO.md) ā€“ Works without internet! **Deployment**: -- [`PROXMOX_QUICKSTART.md`](PROXMOX_QUICKSTART.md) ā€“ ā­ **Proxmox quick deployment (5 min)** -- [`AUTOMATION_QUICKSTART.md`](AUTOMATION_QUICKSTART.md) ā€“ ā­ **Automated reporting setup (5 min)** +- [`PROXMOX_QUICKSTART.md`](docs/PROXMOX_QUICKSTART.md) ā€“ ā­ **Proxmox quick deployment (5 min)** +- [`AUTOMATION_QUICKSTART.md`](docs/AUTOMATION_QUICKSTART.md) ā€“ ā­ **Automated reporting setup (5 min)** - [`docs/07_deployment.md`](docs/07_deployment.md) ā€“ Full deployment guide (all platforms) - [`docs/08_proxmox_deployment.md`](docs/08_proxmox_deployment.md) ā€“ Proxmox detailed guide - [`docs/12_automation_and_reporting.md`](docs/12_automation_and_reporting.md) ā€“ Automation & CI/CD guide @@ -187,7 +187,7 @@ POTE now includes a complete 3-phase monitoring system: - Analyzes by ticker, sector, and political party - Generates comprehensive reports -**Full Documentation**: See [`MONITORING_SYSTEM_COMPLETE.md`](MONITORING_SYSTEM_COMPLETE.md) +**Full Documentation**: See [`MONITORING_SYSTEM_COMPLETE.md`](docs/MONITORING_SYSTEM_COMPLETE.md) ## Next Steps diff --git a/AUTOMATION_QUICKSTART.md b/docs/AUTOMATION_QUICKSTART.md similarity index 100% rename from AUTOMATION_QUICKSTART.md rename to docs/AUTOMATION_QUICKSTART.md diff --git a/DEPLOYMENT_AND_AUTOMATION.md b/docs/DEPLOYMENT_AND_AUTOMATION.md similarity index 100% rename from DEPLOYMENT_AND_AUTOMATION.md rename to docs/DEPLOYMENT_AND_AUTOMATION.md diff --git a/EMAIL_SETUP.md b/docs/EMAIL_SETUP.md similarity index 100% rename from EMAIL_SETUP.md rename to docs/EMAIL_SETUP.md diff --git a/FREE_TESTING_QUICKSTART.md b/docs/FREE_TESTING_QUICKSTART.md similarity index 100% rename from FREE_TESTING_QUICKSTART.md rename to docs/FREE_TESTING_QUICKSTART.md diff --git a/GITEA_SECRETS_GUIDE.md b/docs/GITEA_SECRETS_GUIDE.md similarity index 100% rename from GITEA_SECRETS_GUIDE.md rename to docs/GITEA_SECRETS_GUIDE.md diff --git a/docs/HANDOFF-2026-05-27.md b/docs/HANDOFF-2026-05-27.md index b101663..7deb440 100644 --- a/docs/HANDOFF-2026-05-27.md +++ b/docs/HANDOFF-2026-05-27.md @@ -150,9 +150,9 @@ make beszel-install-agents BESZEL_ONLY=pote-236 # if agent not yet installed | Doc | Purpose | |-----|---------| -| [EMAIL_SETUP.md](../EMAIL_SETUP.md) | SMTP / Mailcow / levkine.ca | -| [AUTOMATION_QUICKSTART.md](../AUTOMATION_QUICKSTART.md) | Cron + reports | -| [PROXMOX_QUICKSTART.md](../PROXMOX_QUICKSTART.md) | Original LXC provisioning | +| [EMAIL_SETUP.md](EMAIL_SETUP.md) | SMTP / Mailcow / levkine.ca | +| [AUTOMATION_QUICKSTART.md](AUTOMATION_QUICKSTART.md) | Cron + reports | +| [PROXMOX_QUICKSTART.md](PROXMOX_QUICKSTART.md) | Original LXC provisioning | | Ansible `docs/guides/projects-handoff-2026-05-26.md` | Multi-project homelab context | | Ansible `docs/guides/smtp-inventory.md` | Mailboxes | diff --git a/LOCAL_TEST_GUIDE.md b/docs/LOCAL_TEST_GUIDE.md similarity index 100% rename from LOCAL_TEST_GUIDE.md rename to docs/LOCAL_TEST_GUIDE.md diff --git a/MONITORING_SYSTEM_COMPLETE.md b/docs/MONITORING_SYSTEM_COMPLETE.md similarity index 100% rename from MONITORING_SYSTEM_COMPLETE.md rename to docs/MONITORING_SYSTEM_COMPLETE.md diff --git a/OFFLINE_DEMO.md b/docs/OFFLINE_DEMO.md similarity index 100% rename from OFFLINE_DEMO.md rename to docs/OFFLINE_DEMO.md diff --git a/PROXMOX_QUICKSTART.md b/docs/PROXMOX_QUICKSTART.md similarity index 100% rename from PROXMOX_QUICKSTART.md rename to docs/PROXMOX_QUICKSTART.md diff --git a/QUICKSTART.md b/docs/QUICKSTART.md similarity index 100% rename from QUICKSTART.md rename to docs/QUICKSTART.md diff --git a/QUICK_SETUP_CARD.md b/docs/QUICK_SETUP_CARD.md similarity index 100% rename from QUICK_SETUP_CARD.md rename to docs/QUICK_SETUP_CARD.md diff --git a/STATUS.md b/docs/STATUS.md similarity index 100% rename from STATUS.md rename to docs/STATUS.md diff --git a/TESTING_STATUS.md b/docs/TESTING_STATUS.md similarity index 100% rename from TESTING_STATUS.md rename to docs/TESTING_STATUS.md diff --git a/WATCHLIST_GUIDE.md b/docs/WATCHLIST_GUIDE.md similarity index 100% rename from WATCHLIST_GUIDE.md rename to docs/WATCHLIST_GUIDE.md diff --git a/pyproject.toml b/pyproject.toml index 8259445..0cb7c18 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,6 +41,8 @@ where = ["src"] [tool.ruff] line-length = 100 target-version = "py311" + +[tool.ruff.lint] select = ["E", "F", "W", "I", "N", "UP", "B", "A", "C4", "SIM", "RET"] ignore = ["E501"] # Line too long (handled by black) diff --git a/scripts/monitor_market.py b/scripts/monitor_market.py index 9bcca90..29df27e 100755 --- a/scripts/monitor_market.py +++ b/scripts/monitor_market.py @@ -74,7 +74,7 @@ def main(tickers, interval, once, min_severity, save_report, lookback): if filtered: # Generate report - report = alert_mgr.generate_summary_report(filtered, format="text") + report = alert_mgr.generate_summary_report(filtered, output_format="text") print("\n" + report) # Save report if requested diff --git a/src/pote/analytics/__init__.py b/src/pote/analytics/__init__.py index d2639b7..4b56ed6 100644 --- a/src/pote/analytics/__init__.py +++ b/src/pote/analytics/__init__.py @@ -2,14 +2,12 @@ Analytics module for calculating returns, performance metrics, and signals. """ -from .returns import ReturnCalculator from .benchmarks import BenchmarkComparison from .metrics import PerformanceMetrics +from .returns import ReturnCalculator __all__ = [ "ReturnCalculator", "BenchmarkComparison", "PerformanceMetrics", ] - - diff --git a/src/pote/analytics/benchmarks.py b/src/pote/analytics/benchmarks.py index 568904d..23959e2 100644 --- a/src/pote/analytics/benchmarks.py +++ b/src/pote/analytics/benchmarks.py @@ -3,7 +3,7 @@ Benchmark comparison for calculating abnormal returns (alpha). """ import logging -from datetime import date, timedelta +from datetime import date from decimal import Decimal from sqlalchemy.orm import Session @@ -60,8 +60,7 @@ class BenchmarkComparison: return None # Calculate return - return_pct = ((end_price - start_price) / start_price) * 100 - return return_pct + return ((end_price - start_price) / start_price) * 100 def calculate_abnormal_return( self, @@ -219,5 +218,3 @@ class BenchmarkComparison: "benchmark": self.BENCHMARKS.get(benchmark, benchmark), "window_days": window_days, } - - diff --git a/src/pote/analytics/metrics.py b/src/pote/analytics/metrics.py index dca08df..db71bbb 100644 --- a/src/pote/analytics/metrics.py +++ b/src/pote/analytics/metrics.py @@ -4,7 +4,6 @@ Performance metrics and aggregations. import logging from collections import defaultdict -from datetime import date from sqlalchemy import func from sqlalchemy.orm import Session @@ -52,11 +51,7 @@ class PerformanceMetrics: if not official: return {"error": "Official not found"} - trades = ( - self.session.query(Trade) - .filter(Trade.official_id == official_id) - .all() - ) + trades = self.session.query(Trade).filter(Trade.official_id == official_id).all() if not trades: return { @@ -70,9 +65,7 @@ class PerformanceMetrics: # Calculate returns for all trades returns_data = [] for trade in trades: - result = self.benchmark.compare_trade_to_benchmark( - trade, window_days, benchmark - ) + result = self.benchmark.compare_trade_to_benchmark(trade, window_days, benchmark) if result: returns_data.append(result) @@ -96,9 +89,7 @@ class PerformanceMetrics: worst_trade = min(returns_data, key=lambda x: x["trade_return"]) # Total value traded - total_value = sum( - float(t.value_min or 0) for t in trades if t.value_min - ) + total_value = sum(float(t.value_min or 0) for t in trades if t.value_min) return { "name": official.name, @@ -154,20 +145,14 @@ class PerformanceMetrics: List of sector performance dictionaries """ # Get all trades with security info - trades = ( - self.session.query(Trade) - .join(Security) - .all() - ) + trades = self.session.query(Trade).join(Security).all() # Group by sector sector_data = defaultdict(list) for trade in trades: sector = trade.security.sector or "Unknown" - result = self.benchmark.compare_trade_to_benchmark( - trade, window_days, benchmark - ) + result = self.benchmark.compare_trade_to_benchmark(trade, window_days, benchmark) if result: sector_data[sector].append(result) @@ -180,14 +165,16 @@ class PerformanceMetrics: returns = [d["trade_return"] for d in data] alphas = [d["abnormal_return"] for d in data] - results.append({ - "sector": sector, - "trade_count": len(data), - "avg_return": sum(returns) / len(returns), - "avg_alpha": sum(alphas) / len(alphas), - "win_rate": sum(1 for r in returns if r > 0) / len(returns), - "beat_market_rate": sum(1 for a in alphas if a > 0) / len(alphas), - }) + results.append( + { + "sector": sector, + "trade_count": len(data), + "avg_return": sum(returns) / len(returns), + "avg_alpha": sum(alphas) / len(alphas), + "win_rate": sum(1 for r in returns if r > 0) / len(returns), + "beat_market_rate": sum(1 for a in alphas if a > 0) / len(alphas), + } + ) # Sort by average alpha results.sort(key=lambda x: x["avg_alpha"], reverse=True) @@ -229,11 +216,7 @@ class PerformanceMetrics: Returns: Dictionary with timing statistics """ - trades = ( - self.session.query(Trade) - .filter(Trade.filing_date.isnot(None)) - .all() - ) + trades = self.session.query(Trade).filter(Trade.filing_date.isnot(None)).all() if not trades: return {"error": "No trades with disclosure dates"} @@ -288,5 +271,3 @@ class PerformanceMetrics: "benchmark": benchmark, **aggregate, } - - diff --git a/src/pote/analytics/returns.py b/src/pote/analytics/returns.py index 75d45d1..4e9701b 100644 --- a/src/pote/analytics/returns.py +++ b/src/pote/analytics/returns.py @@ -94,18 +94,20 @@ class ReturnCalculator: def calculate_multiple_windows( self, trade: Trade, - windows: list[int] = [30, 60, 90, 180], + windows: list[int] | None = None, ) -> dict[int, dict]: """ Calculate returns for multiple time windows. Args: trade: Trade object - windows: List of window sizes in days + windows: List of window sizes in days (defaults to 30/60/90/180) Returns: Dictionary mapping window_days to return metrics """ + if windows is None: + windows = [30, 60, 90, 180] results = {} for window in windows: result = self.calculate_trade_return(trade, window) @@ -223,12 +225,13 @@ class ReturnCalculator: if not prices: return pd.DataFrame() + # open/high/low are nullable columns; use NaN (pandas-native) when absent. data = [ { "date": p.date, - "open": float(p.open), - "high": float(p.high), - "low": float(p.low), + "open": float(p.open) if p.open is not None else float("nan"), + "high": float(p.high) if p.high is not None else float("nan"), + "low": float(p.low) if p.low is not None else float("nan"), "close": float(p.close), "volume": p.volume, } @@ -236,4 +239,3 @@ class ReturnCalculator: ] return pd.DataFrame(data) - diff --git a/src/pote/db/models.py b/src/pote/db/models.py index 03739c0..5d90dc3 100644 --- a/src/pote/db/models.py +++ b/src/pote/db/models.py @@ -3,17 +3,17 @@ SQLAlchemy ORM models for POTE. Matches the schema defined in docs/02_data_model.md. """ -from datetime import date, datetime, timezone +from datetime import UTC, date, datetime from decimal import Decimal from sqlalchemy import ( DECIMAL, + JSON, Date, DateTime, ForeignKey, Index, Integer, - JSON, String, Text, UniqueConstraint, @@ -35,13 +35,11 @@ class Official(Base): state: Mapped[str | None] = mapped_column(String(2)) bioguide_id: Mapped[str | None] = mapped_column(String(20), unique=True) external_ids: Mapped[str | None] = mapped_column(Text) # JSON blob for other IDs - created_at: Mapped[datetime] = mapped_column( - DateTime, default=lambda: datetime.now(timezone.utc) - ) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(UTC)) updated_at: Mapped[datetime] = mapped_column( DateTime, - default=lambda: datetime.now(timezone.utc), - onupdate=lambda: datetime.now(timezone.utc), + default=lambda: datetime.now(UTC), + onupdate=lambda: datetime.now(UTC), ) # Relationships @@ -63,13 +61,11 @@ class Security(Base): sector: Mapped[str | None] = mapped_column(String(100)) industry: Mapped[str | None] = mapped_column(String(100)) asset_type: Mapped[str] = mapped_column(String(50), default="stock") # stock, bond, etc. - created_at: Mapped[datetime] = mapped_column( - DateTime, default=lambda: datetime.now(timezone.utc) - ) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(UTC)) updated_at: Mapped[datetime] = mapped_column( DateTime, - default=lambda: datetime.now(timezone.utc), - onupdate=lambda: datetime.now(timezone.utc), + default=lambda: datetime.now(UTC), + onupdate=lambda: datetime.now(UTC), ) # Relationships @@ -107,13 +103,11 @@ class Trade(Base): # Quality flags (JSON or enum list) quality_flags: Mapped[str | None] = mapped_column(Text) # e.g., "range_only,delayed_filing" - created_at: Mapped[datetime] = mapped_column( - DateTime, default=lambda: datetime.now(timezone.utc) - ) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(UTC)) updated_at: Mapped[datetime] = mapped_column( DateTime, - default=lambda: datetime.now(timezone.utc), - onupdate=lambda: datetime.now(timezone.utc), + default=lambda: datetime.now(UTC), + onupdate=lambda: datetime.now(UTC), ) # Relationships @@ -156,9 +150,7 @@ class Price(Base): adjusted_close: Mapped[Decimal | None] = mapped_column(DECIMAL(15, 4)) source: Mapped[str] = mapped_column(String(50), default="yfinance") - created_at: Mapped[datetime] = mapped_column( - DateTime, default=lambda: datetime.now(timezone.utc) - ) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(UTC)) # Relationships security: Mapped["Security"] = relationship("Security", back_populates="prices") @@ -188,9 +180,7 @@ class MetricOfficial(Base): avg_abnormal_return_1m: Mapped[Decimal | None] = mapped_column(DECIMAL(10, 6)) cluster_label: Mapped[str | None] = mapped_column(String(50)) - created_at: Mapped[datetime] = mapped_column( - DateTime, default=lambda: datetime.now(timezone.utc) - ) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(UTC)) __table_args__ = ( UniqueConstraint("official_id", "calc_date", "calc_version", name="uq_metrics_official"), @@ -212,9 +202,7 @@ class MetricTrade(Base): abnormal_return_1m: Mapped[Decimal | None] = mapped_column(DECIMAL(10, 6)) signal_flags: Mapped[str | None] = mapped_column(Text) # JSON list - created_at: Mapped[datetime] = mapped_column( - DateTime, default=lambda: datetime.now(timezone.utc) - ) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(UTC)) __table_args__ = ( UniqueConstraint("trade_id", "calc_date", "calc_version", name="uq_metrics_trade"), @@ -242,18 +230,14 @@ class MarketAlert(Base): # Metrics at time of alert price: Mapped[Decimal | None] = mapped_column(DECIMAL(15, 4)) volume: Mapped[int | None] = mapped_column(Integer) - change_pct: Mapped[Decimal | None] = mapped_column( - DECIMAL(10, 4) - ) # Price change % + change_pct: Mapped[Decimal | None] = mapped_column(DECIMAL(10, 4)) # Price change % # Severity scoring severity: Mapped[int | None] = mapped_column(Integer) # 1-10 scale # Metadata source: Mapped[str] = mapped_column(String(50), default="market_monitor") - created_at: Mapped[datetime] = mapped_column( - DateTime, default=lambda: datetime.now(timezone.utc) - ) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(UTC)) # Indexes for efficient queries __table_args__ = ( diff --git a/src/pote/ingestion/house_watcher.py b/src/pote/ingestion/house_watcher.py index ee5e745..ab3880a 100644 --- a/src/pote/ingestion/house_watcher.py +++ b/src/pote/ingestion/house_watcher.py @@ -14,6 +14,7 @@ import httpx logger = logging.getLogger(__name__) + def _default_data_urls() -> tuple[str, ...]: override = os.environ.get("POTE_HOUSE_DATA_URL", "").strip() if override: @@ -235,10 +236,9 @@ def normalize_transaction_type(txn_type: str) -> str: if "purchase" in txn_lower or "buy" in txn_lower: return "buy" - elif "sale" in txn_lower or "sell" in txn_lower: + if "sale" in txn_lower or "sell" in txn_lower: return "sell" - elif "exchange" in txn_lower: + if "exchange" in txn_lower: return "exchange" - else: - # Default to the original, lowercased - return txn_lower + # Default to the original, lowercased + return txn_lower diff --git a/src/pote/ingestion/prices.py b/src/pote/ingestion/prices.py index 56345a2..463e351 100644 --- a/src/pote/ingestion/prices.py +++ b/src/pote/ingestion/prices.py @@ -4,7 +4,7 @@ Fetches daily OHLCV data for securities and stores in the prices table. """ import logging -from datetime import date, datetime, timedelta, timezone +from datetime import UTC, date, datetime, timedelta from decimal import Decimal import pandas as pd @@ -158,7 +158,7 @@ class PriceLoader: "volume": int(row["volume"]) if pd.notna(row.get("volume")) else None, "adjusted_close": None, # We'll compute this later if needed "source": "yfinance", - "created_at": datetime.now(timezone.utc), + "created_at": datetime.now(UTC), } records.append(record) diff --git a/src/pote/monitoring/__init__.py b/src/pote/monitoring/__init__.py index 4c794a4..7205f97 100644 --- a/src/pote/monitoring/__init__.py +++ b/src/pote/monitoring/__init__.py @@ -9,4 +9,3 @@ from .market_monitor import MarketMonitor from .pattern_detector import PatternDetector __all__ = ["MarketMonitor", "AlertManager", "DisclosureCorrelator", "PatternDetector"] - diff --git a/src/pote/monitoring/alert_manager.py b/src/pote/monitoring/alert_manager.py index c56117e..1f36c52 100644 --- a/src/pote/monitoring/alert_manager.py +++ b/src/pote/monitoring/alert_manager.py @@ -4,8 +4,7 @@ Handles alert filtering, formatting, and delivery. """ import logging -from datetime import datetime, timezone -from typing import Any +from datetime import UTC, datetime from sqlalchemy.orm import Session @@ -75,21 +74,24 @@ class AlertManager: Returns: HTML formatted alert """ - severity_class = "high" if (alert.severity or 0) >= 7 else "medium" if (alert.severity or 0) >= 4 else "low" + severity_class = ( + "high" + if (alert.severity or 0) >= 7 + else "medium" if (alert.severity or 0) >= 4 else "low" + ) - html = f""" + return f"""

{alert.ticker} - {alert.alert_type.replace('_', ' ').title()}

{alert.timestamp.strftime('%Y-%m-%d %H:%M:%S')}

Severity: {alert.severity}/10

- Price: ${float(alert.price):.2f} + Price: ${float(alert.price or 0):.2f} Volume: {alert.volume:,} - Change: {float(alert.change_pct):+.2f}% + Change: {float(alert.change_pct or 0):+.2f}%
""" - return html def filter_alerts( self, @@ -117,7 +119,7 @@ class AlertManager: # Filter by ticker if tickers: - ticker_set = set(t.upper() for t in tickers) + ticker_set = {t.upper() for t in tickers} filtered = [a for a in filtered if a.ticker.upper() in ticker_set] # Filter by alert type @@ -128,22 +130,21 @@ class AlertManager: return filtered def generate_summary_report( - self, alerts: list[MarketAlert], format: str = "text" + self, alerts: list[MarketAlert], output_format: str = "text" ) -> str: """ Generate summary report of alerts. Args: alerts: List of alerts - format: Output format ('text' or 'html') + output_format: Output format ('text' or 'html') Returns: Formatted summary report """ - if format == "html": + if output_format == "html": return self._generate_html_summary(alerts) - else: - return self._generate_text_summary(alerts) + return self._generate_text_summary(alerts) def _generate_text_summary(self, alerts: list[MarketAlert]) -> str: """Generate text summary report.""" @@ -152,7 +153,7 @@ class AlertManager: lines = [ "=" * 80, - f" MARKET ACTIVITY ALERTS - {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')} UTC", + f" MARKET ACTIVITY ALERTS - {datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S')} UTC", f" {len(alerts)} Alerts", "=" * 80, "", @@ -180,9 +181,7 @@ class AlertManager: lines.append(f"šŸŽÆ {ticker} - {len(ticker_alerts)} alerts (Max Severity: {max_sev}/10)") lines.append("ā”€" * 80) - for alert in sorted( - ticker_alerts, key=lambda a: a.severity or 0, reverse=True - ): + for alert in sorted(ticker_alerts, key=lambda a: a.severity or 0, reverse=True): lines.append("") lines.append(self.format_alert_text(alert)) @@ -202,9 +201,7 @@ class AlertManager: type_counts[alert.alert_type] = type_counts.get(alert.alert_type, 0) + 1 lines.append("\nAlert Types:") - for alert_type, count in sorted( - type_counts.items(), key=lambda x: x[1], reverse=True - ): + for alert_type, count in sorted(type_counts.items(), key=lambda x: x[1], reverse=True): lines.append(f" {alert_type.replace('_', ' ').title():20s}: {count}") # Top severity alerts @@ -232,8 +229,8 @@ class AlertManager: ".timestamp { color: #666; font-size: 0.9em; }", ".metrics span { margin-right: 20px; }", "", - f"

Market Activity Alerts

", - f"

{len(alerts)} Alerts | {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')} UTC

", + "

Market Activity Alerts

", + f"

{len(alerts)} Alerts | {datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S')} UTC

", ] for alert in sorted(alerts, key=lambda a: a.severity or 0, reverse=True): @@ -241,5 +238,3 @@ class AlertManager: html_parts.append("") return "\n".join(html_parts) - - diff --git a/src/pote/monitoring/disclosure_correlator.py b/src/pote/monitoring/disclosure_correlator.py index d13bcfa..bf0d34e 100644 --- a/src/pote/monitoring/disclosure_correlator.py +++ b/src/pote/monitoring/disclosure_correlator.py @@ -5,14 +5,13 @@ Calculates timing advantage and suspicious activity scores. """ import logging -from datetime import date, timedelta, timezone -from decimal import Decimal +from datetime import UTC, date, timedelta from typing import Any -from sqlalchemy import and_, func +from sqlalchemy import and_ from sqlalchemy.orm import Session -from pote.db.models import MarketAlert, Official, Security, Trade +from pote.db.models import MarketAlert, Security, Trade logger = logging.getLogger(__name__) @@ -27,9 +26,7 @@ class DisclosureCorrelator: """Initialize disclosure correlator.""" self.session = session - def get_alerts_before_trade( - self, trade: Trade, lookback_days: int = 30 - ) -> list[MarketAlert]: + def get_alerts_before_trade(self, trade: Trade, lookback_days: int = 30) -> list[MarketAlert]: """ Get market alerts that occurred BEFORE a trade. @@ -50,14 +47,10 @@ class DisclosureCorrelator: # Convert dates to datetime for comparison from datetime import datetime - start_dt = datetime.combine(start_date, datetime.min.time()).replace( - tzinfo=timezone.utc - ) - end_dt = datetime.combine(end_date, datetime.max.time()).replace( - tzinfo=timezone.utc - ) + start_dt = datetime.combine(start_date, datetime.min.time()).replace(tzinfo=UTC) + end_dt = datetime.combine(end_date, datetime.max.time()).replace(tzinfo=UTC) - alerts = ( + return ( self.session.query(MarketAlert) .filter( and_( @@ -70,8 +63,6 @@ class DisclosureCorrelator: .all() ) - return alerts - def calculate_timing_score( self, trade: Trade, prior_alerts: list[MarketAlert] ) -> dict[str, Any]: @@ -131,8 +122,7 @@ class DisclosureCorrelator: ) elif suspicious: reason = ( - f"Trade occurred after {len(prior_alerts)} alerts. " - f"Possible timing advantage." + f"Trade occurred after {len(prior_alerts)} alerts. " f"Possible timing advantage." ) else: reason = ( @@ -170,35 +160,27 @@ class DisclosureCorrelator: timing_analysis = self.calculate_timing_score(trade, prior_alerts) # Build full analysis - analysis = { + return { "trade_id": trade.id, "official_name": trade.official.name if trade.official else None, "ticker": trade.security.ticker if trade.security else None, "side": trade.side, "transaction_date": str(trade.transaction_date), "filing_date": str(trade.filing_date) if trade.filing_date else None, - "value_range": f"${float(trade.value_min):,.0f}" - + ( - f"-${float(trade.value_max):,.0f}" - if trade.value_max - else "+" - ), + "value_range": f"${float(trade.value_min or 0):,.0f}" + + (f"-${float(trade.value_max):,.0f}" if trade.value_max else "+"), **timing_analysis, "prior_alerts": [ { "timestamp": str(alert.timestamp), "alert_type": alert.alert_type, "severity": alert.severity, - "days_before_trade": ( - trade.transaction_date - alert.timestamp.date() - ).days, + "days_before_trade": (trade.transaction_date - alert.timestamp.date()).days, } for alert in prior_alerts ], } - return analysis - def analyze_recent_disclosures( self, days: int = 7, min_timing_score: float = 50 ) -> list[dict[str, Any]]: @@ -237,9 +219,7 @@ class DisclosureCorrelator: f"Found {len(suspicious_trades)} trades with timing score >= {min_timing_score}" ) - return sorted( - suspicious_trades, key=lambda x: x["timing_score"], reverse=True - ) + return sorted(suspicious_trades, key=lambda x: x["timing_score"], reverse=True) def get_official_timing_pattern( self, official_id: int, lookback_days: int = 365 @@ -258,9 +238,7 @@ class DisclosureCorrelator: trades = ( self.session.query(Trade) - .filter( - and_(Trade.official_id == official_id, Trade.transaction_date >= since_date) - ) + .filter(and_(Trade.official_id == official_id, Trade.transaction_date >= since_date)) .join(Trade.security) .all() ) @@ -285,9 +263,7 @@ class DisclosureCorrelator: highly_suspicious = sum(1 for a in analyses if a.get("highly_suspicious", False)) avg_timing_score = ( - sum(a["timing_score"] for a in analyses) / total_trades - if total_trades > 0 - else 0 + sum(a["timing_score"] for a in analyses) / total_trades if total_trades > 0 else 0 ) # Determine pattern @@ -311,9 +287,7 @@ class DisclosureCorrelator: "analyses": analyses, } - def get_ticker_timing_analysis( - self, ticker: str, lookback_days: int = 365 - ) -> dict[str, Any]: + def get_ticker_timing_analysis(self, ticker: str, lookback_days: int = 365) -> dict[str, Any]: """ Analyze timing patterns for a specific ticker. @@ -329,9 +303,7 @@ class DisclosureCorrelator: trades = ( self.session.query(Trade) .join(Trade.security) - .filter( - and_(Security.ticker == ticker, Trade.transaction_date >= since_date) - ) + .filter(and_(Security.ticker == ticker, Trade.transaction_date >= since_date)) .join(Trade.official) .all() ) @@ -350,10 +322,6 @@ class DisclosureCorrelator: "trade_count": len(analyses), "trades_with_alerts": sum(1 for a in analyses if a["alert_count"] > 0), "suspicious_count": sum(1 for a in analyses if a["suspicious"]), - "avg_timing_score": round( - sum(a["timing_score"] for a in analyses) / len(analyses), 2 - ), + "avg_timing_score": round(sum(a["timing_score"] for a in analyses) / len(analyses), 2), "analyses": sorted(analyses, key=lambda x: x["timing_score"], reverse=True), } - - diff --git a/src/pote/monitoring/market_monitor.py b/src/pote/monitoring/market_monitor.py index 1f98ace..abe6333 100644 --- a/src/pote/monitoring/market_monitor.py +++ b/src/pote/monitoring/market_monitor.py @@ -4,7 +4,7 @@ Detects unusual activity: volume spikes, price movements, volatility. """ import logging -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from decimal import Decimal from typing import Any @@ -59,7 +59,7 @@ class MarketMonitor: Returns: List of alerts detected """ - alerts = [] + alerts: list[dict[str, Any]] = [] try: stock = yf.Ticker(ticker) @@ -90,7 +90,7 @@ class MarketMonitor: { "ticker": ticker, "alert_type": "unusual_volume", - "timestamp": datetime.now(timezone.utc), + "timestamp": datetime.now(UTC), "details": { "current_volume": int(current_volume), "avg_volume": int(avg_volume), @@ -109,10 +109,8 @@ class MarketMonitor: alerts.append( { "ticker": ticker, - "alert_type": "price_spike" - if price_change > 0 - else "price_drop", - "timestamp": datetime.now(timezone.utc), + "alert_type": "price_spike" if price_change > 0 else "price_drop", + "timestamp": datetime.now(UTC), "details": { "current_price": float(current_price), "prev_price": float(prev["Close"]), @@ -129,14 +127,12 @@ class MarketMonitor: if len(hist) >= 5: recent_volatility = hist["Close"].iloc[-5:].pct_change().abs().mean() if recent_volatility > avg_price_change * 2 and avg_price_change > 0: - severity = min( - 10, int((recent_volatility / avg_price_change) - 1) - ) + severity = min(10, int((recent_volatility / avg_price_change) - 1)) alerts.append( { "ticker": ticker, "alert_type": "high_volatility", - "timestamp": datetime.now(timezone.utc), + "timestamp": datetime.now(UTC), "details": { "recent_volatility": round(recent_volatility * 100, 2), "avg_volatility": round(avg_price_change * 100, 2), @@ -227,7 +223,7 @@ class MarketMonitor: Returns: List of MarketAlert objects """ - since = datetime.now(timezone.utc) - timedelta(days=days) + since = datetime.now(UTC) - timedelta(days=days) query = self.session.query(MarketAlert).filter(MarketAlert.timestamp >= since) @@ -252,7 +248,7 @@ class MarketMonitor: Returns: Dict mapping ticker to alert summary """ - since = datetime.now(timezone.utc) - timedelta(days=days) + since = datetime.now(UTC) - timedelta(days=days) from sqlalchemy import func @@ -278,5 +274,3 @@ class MarketMonitor: } return summary - - diff --git a/src/pote/monitoring/pattern_detector.py b/src/pote/monitoring/pattern_detector.py index e92bbf7..66bba6a 100644 --- a/src/pote/monitoring/pattern_detector.py +++ b/src/pote/monitoring/pattern_detector.py @@ -5,13 +5,12 @@ Identifies recurring suspicious behavior and trading patterns. import logging from datetime import date, timedelta -from decimal import Decimal from typing import Any -from sqlalchemy import and_, func +from sqlalchemy import func from sqlalchemy.orm import Session -from pote.db.models import MarketAlert, Official, Security, Trade +from pote.db.models import Official, Security, Trade from pote.monitoring.disclosure_correlator import DisclosureCorrelator logger = logging.getLogger(__name__) @@ -60,9 +59,7 @@ class PatternDetector: .all() ) - logger.info( - f"Analyzing {len(officials_with_trades)} officials with {min_trades}+ trades" - ) + logger.info(f"Analyzing {len(officials_with_trades)} officials with {min_trades}+ trades") rankings = [] @@ -70,9 +67,7 @@ class PatternDetector: official_id, name, chamber, party, state, trade_count = official_data # Get timing pattern - pattern = self.correlator.get_official_timing_pattern( - official_id, lookback_days - ) + pattern = self.correlator.get_official_timing_pattern(official_id, lookback_days) if pattern["trade_count"] == 0: continue @@ -128,9 +123,7 @@ class PatternDetector: rankings = self.rank_officials_by_timing(lookback_days, min_trades=5) # Filter for high suspicious rates - offenders = [ - r for r in rankings if r["suspicious_rate"] >= min_suspicious_rate * 100 - ] + offenders = [r for r in rankings if r["suspicious_rate"] >= min_suspicious_rate * 100] logger.info( f"Found {len(offenders)} officials with {min_suspicious_rate*100}%+ suspicious trades" @@ -155,9 +148,7 @@ class PatternDetector: # Get tickers with enough trades tickers_with_trades = ( - self.session.query( - Security.ticker, func.count(Trade.id).label("trade_count") - ) + self.session.query(Security.ticker, func.count(Trade.id).label("trade_count")) .join(Trade) .filter(Trade.transaction_date >= since_date) .group_by(Security.ticker) @@ -169,10 +160,8 @@ class PatternDetector: ticker_patterns = [] - for ticker, trade_count in tickers_with_trades: - analysis = self.correlator.get_ticker_timing_analysis( - ticker, lookback_days - ) + for ticker, _trade_count in tickers_with_trades: + analysis = self.correlator.get_ticker_timing_analysis(ticker, lookback_days) if analysis["trade_count"] == 0: continue @@ -199,9 +188,7 @@ class PatternDetector: return ticker_patterns - def get_sector_timing_analysis( - self, lookback_days: int = 365 - ) -> dict[str, dict[str, Any]]: + def get_sector_timing_analysis(self, lookback_days: int = 365) -> dict[str, dict[str, Any]]: """ Analyze timing patterns by sector. @@ -252,7 +239,7 @@ class PatternDetector: sector_stats[sector]["suspicious_count"] += 1 # Calculate averages - for sector, stats in sector_stats.items(): + for stats in sector_stats.values(): if stats["trade_count"] > 0: stats["avg_timing_score"] = round( stats["total_timing_score"] / stats["trade_count"], 2 @@ -266,9 +253,7 @@ class PatternDetector: return sector_stats - def get_party_comparison( - self, lookback_days: int = 365 - ) -> dict[str, dict[str, Any]]: + def get_party_comparison(self, lookback_days: int = 365) -> dict[str, dict[str, Any]]: """ Compare timing patterns between political parties. @@ -303,7 +288,7 @@ class PatternDetector: party_stats[party]["officials"].append(ranking) # Calculate averages - for party, stats in party_stats.items(): + for stats in party_stats.values(): if stats["total_trades"] > 0: stats["avg_timing_score"] = round( stats["total_timing_score"] / stats["total_trades"], 2 @@ -336,7 +321,7 @@ class PatternDetector: # Calculate summary statistics total_officials = len(official_rankings) total_offenders = len(repeat_offenders) - + avg_timing_score = ( sum(r["avg_timing_score"] for r in official_rankings) / total_officials if total_officials > 0 @@ -356,5 +341,3 @@ class PatternDetector: "sector_analysis": sector_analysis, "party_comparison": party_comparison, } - - diff --git a/src/pote/reporting/__init__.py b/src/pote/reporting/__init__.py index 3d56b64..771ea69 100644 --- a/src/pote/reporting/__init__.py +++ b/src/pote/reporting/__init__.py @@ -8,5 +8,3 @@ from .email_reporter import EmailReporter from .report_generator import ReportGenerator __all__ = ["EmailReporter", "ReportGenerator"] - - diff --git a/src/pote/reporting/email_reporter.py b/src/pote/reporting/email_reporter.py index 8cfae2a..c515e22 100644 --- a/src/pote/reporting/email_reporter.py +++ b/src/pote/reporting/email_reporter.py @@ -8,7 +8,6 @@ import logging import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText -from typing import List, Optional from pote.config import settings @@ -20,31 +19,30 @@ class EmailReporter: def __init__( self, - smtp_host: Optional[str] = None, - smtp_port: Optional[int] = None, - smtp_user: Optional[str] = None, - smtp_password: Optional[str] = None, - from_email: Optional[str] = None, + smtp_host: str | None = None, + smtp_port: int | None = None, + smtp_user: str | None = None, + smtp_password: str | None = None, + from_email: str | None = None, ): """ Initialize email reporter. If parameters are not provided, will attempt to use settings from config. """ - self.smtp_host = smtp_host or getattr(settings, "smtp_host", "localhost") - self.smtp_port = smtp_port or getattr(settings, "smtp_port", 587) - self.smtp_user = smtp_user or getattr(settings, "smtp_user", None) - self.smtp_password = smtp_password or getattr(settings, "smtp_password", None) - self.from_email = from_email or getattr( - settings, "from_email", "pote@localhost" - ) + # Settings always defines these fields (with defaults), so direct access is safe. + self.smtp_host: str = smtp_host or settings.smtp_host + self.smtp_port: int = smtp_port or settings.smtp_port + self.smtp_user: str = smtp_user or settings.smtp_user + self.smtp_password: str = smtp_password or settings.smtp_password + self.from_email: str = from_email or settings.from_email or "pote@localhost" def send_report( self, - to_emails: List[str], + to_emails: list[str], subject: str, body_text: str, - body_html: Optional[str] = None, + body_html: str | None = None, ) -> bool: """ Send an email report. @@ -113,4 +111,3 @@ class EmailReporter: except Exception as e: logger.error(f"SMTP connection test failed: {e}") return False - diff --git a/src/pote/reporting/report_generator.py b/src/pote/reporting/report_generator.py index 096f3cc..34e86c4 100644 --- a/src/pote/reporting/report_generator.py +++ b/src/pote/reporting/report_generator.py @@ -6,7 +6,7 @@ Generates formatted reports from database data. import logging from datetime import date, datetime, timedelta -from typing import Any, Dict, List, Optional +from typing import Any from sqlalchemy import func from sqlalchemy.orm import Session @@ -27,8 +27,8 @@ class ReportGenerator: self.detector = PatternDetector(session) def generate_daily_summary( - self, report_date: Optional[date] = None, *, lookback_days: int = 1 - ) -> Dict[str, Any]: + self, report_date: date | None = None, *, lookback_days: int = 1 + ) -> dict[str, Any]: """ Generate a daily summary report. @@ -69,7 +69,7 @@ class ReportGenerator: ) # Get high-severity alerts - critical_alerts = [a for a in new_alerts if a.severity >= 7] + critical_alerts = [a for a in new_alerts if (a.severity or 0) >= 7] # Get suspicious timing matches suspicious_trades = [] @@ -110,7 +110,7 @@ class ReportGenerator: "suspicious_trades": suspicious_trades, } - def generate_weekly_summary(self) -> Dict[str, Any]: + def generate_weekly_summary(self) -> dict[str, Any]: """ Generate a weekly summary report. @@ -121,9 +121,7 @@ class ReportGenerator: # Most active officials active_officials = ( - self.session.query( - Official.name, func.count(Trade.id).label("trade_count") - ) + 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) @@ -134,9 +132,7 @@ class ReportGenerator: # Most traded securities active_securities = ( - self.session.query( - Security.ticker, func.count(Trade.id).label("trade_count") - ) + 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) @@ -146,8 +142,10 @@ class ReportGenerator: ) # Get top suspicious patterns + # Same fix as branch docs/deploy-email-closed: the old kwarg names never + # existed on identify_repeat_offenders and failed every weekly run. repeat_offenders = self.detector.identify_repeat_offenders( - days_lookback=7, min_suspicious_trades=2, min_timing_score=40 + lookback_days=7, min_suspicious_rate=0.4 ) return { @@ -157,14 +155,13 @@ class ReportGenerator: {"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 + {"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: + def format_as_text(self, report_data: dict[str, Any], report_type: str) -> str: """ Format report data as plain text. @@ -177,12 +174,11 @@ class ReportGenerator: """ if report_type == "daily": return self._format_daily_text(report_data) - elif report_type == "weekly": + if report_type == "weekly": return self._format_weekly_text(report_data) - else: - return str(report_data) + return str(report_data) - def _format_daily_text(self, data: Dict[str, Any]) -> str: + 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 = ( @@ -245,7 +241,7 @@ class ReportGenerator: return "\n".join(lines) - def _format_weekly_text(self, data: Dict[str, Any]) -> str: + def _format_weekly_text(self, data: dict[str, Any]) -> str: """Format weekly report as plain text.""" lines = [ "=" * 70, @@ -264,9 +260,7 @@ class ReportGenerator: lines.append(f" ā€¢ {security['ticker']}: {security['trade_count']} trades") if data["repeat_offenders"]: - lines.extend( - ["", f"āš ļø REPEAT OFFENDERS ({data['repeat_offenders_count']} total)"] - ) + lines.extend(["", f"āš ļø REPEAT OFFENDERS ({data['repeat_offenders_count']} total)"]) for offender in data["repeat_offenders"]: lines.append( f" ā€¢ {offender['official_name']}: " @@ -285,7 +279,7 @@ class ReportGenerator: return "\n".join(lines) - def format_as_html(self, report_data: Dict[str, Any], report_type: str) -> str: + def format_as_html(self, report_data: dict[str, Any], report_type: str) -> str: """ Format report data as HTML. @@ -298,12 +292,11 @@ class ReportGenerator: """ if report_type == "daily": return self._format_daily_html(report_data) - elif report_type == "weekly": + if report_type == "weekly": return self._format_weekly_html(report_data) - else: - return f"
{report_data}
" + return f"
{report_data}
" - def _format_daily_html(self, data: Dict[str, Any]) -> str: + 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):" @@ -327,7 +320,7 @@ class ReportGenerator:

POTE Daily Report - {data['date']}

- +

šŸ“Š Summary

{new_trades_label} {data['new_trades_count']}
@@ -342,7 +335,7 @@ class ReportGenerator: for t in data["new_trades"][:10]: html += f"""
- {t['official']}: {t['side']} {t['ticker']} + {t['official']}: {t['side']} {t['ticker']} (${t['value_min']:,.0f} - ${t['value_max']:,.0f}) on {t['transaction_date']}
""" @@ -352,7 +345,7 @@ class ReportGenerator: for a in data["critical_alerts"][:5]: html += f"""
- {a['ticker']}: {a['type']} (severity {a['severity']}) + {a['ticker']}: {a['type']} (severity {a['severity']}) at {a['timestamp'].strftime('%H:%M:%S')}
""" @@ -377,7 +370,7 @@ class ReportGenerator: return html - def _format_weekly_html(self, data: Dict[str, Any]) -> str: + def _format_weekly_html(self, data: dict[str, Any]) -> str: """Format weekly report as HTML.""" html = f""" @@ -395,7 +388,7 @@ class ReportGenerator:

POTE Weekly Report

Period: {data['period_start']} to {data['period_end']}

- +

šŸ‘„ Most Active Officials

@@ -406,7 +399,7 @@ class ReportGenerator: html += """
OfficialTrade Count
- +

šŸ“ˆ Most Traded Securities

@@ -443,4 +436,3 @@ class ReportGenerator: """ return html - diff --git a/tests/conftest.py b/tests/conftest.py index 250bd8c..ac45530 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -22,8 +22,8 @@ def test_db_session() -> Session: engine = create_engine("sqlite:///:memory:", echo=False) Base.metadata.create_all(engine) - TestSessionLocal = sessionmaker(bind=engine) - session = TestSessionLocal() + session_factory = sessionmaker(bind=engine) + session = session_factory() yield session diff --git a/tests/test_analytics.py b/tests/test_analytics.py index bd1e3f7..2c9294b 100644 --- a/tests/test_analytics.py +++ b/tests/test_analytics.py @@ -1,27 +1,28 @@ """Tests for analytics module.""" -import pytest from datetime import date, timedelta from decimal import Decimal -from pote.analytics.returns import ReturnCalculator +import pytest + from pote.analytics.benchmarks import BenchmarkComparison from pote.analytics.metrics import PerformanceMetrics -from pote.db.models import Official, Security, Trade, Price +from pote.analytics.returns import ReturnCalculator +from pote.db.models import Price, Security, Trade @pytest.fixture def sample_prices(test_db_session, sample_security): """Create sample price data for testing.""" session = test_db_session - + # Add SPY (benchmark) prices spy = Security(ticker="SPY", name="SPDR S&P 500 ETF") session.add(spy) session.flush() - + base_date = date(2024, 1, 1) - + # Create SPY prices for i in range(100): price = Price( @@ -34,7 +35,7 @@ def sample_prices(test_db_session, sample_security): volume=1000000, ) session.add(price) - + # Create prices for sample_security (AAPL) for i in range(100): price = Price( @@ -47,7 +48,7 @@ def sample_prices(test_db_session, sample_security): volume=50000000, ) session.add(price) - + session.commit() return session @@ -80,7 +81,9 @@ def test_return_calculator_basic(test_db_session, sample_official, sample_securi assert "exit_price" in result -def test_return_calculator_sell_trade(test_db_session, sample_official, sample_security, sample_prices): +def test_return_calculator_sell_trade( + test_db_session, sample_official, sample_security, sample_prices +): session = test_db_session """Test return calculation for sell trade.""" trade = Trade( @@ -130,7 +133,7 @@ def test_benchmark_comparison(test_db_session, sample_official, sample_security, """Test benchmark comparison.""" # Create trade and SPY security spy = session.query(Security).filter_by(ticker="SPY").first() - + trade = Trade( official_id=sample_official.id, security_id=spy.id, @@ -154,12 +157,14 @@ def test_benchmark_comparison(test_db_session, sample_official, sample_security, assert "beat_market" in result -def test_performance_metrics_official(test_db_session, sample_official, sample_security, sample_prices): +def test_performance_metrics_official( + test_db_session, sample_official, sample_security, sample_prices +): session = test_db_session """Test official performance metrics.""" # Create multiple trades spy = session.query(Security).filter_by(ticker="SPY").first() - + for i in range(3): trade = Trade( official_id=sample_official.id, @@ -171,7 +176,7 @@ def test_performance_metrics_official(test_db_session, sample_official, sample_s value_max=Decimal("50000"), ) session.add(trade) - + session.commit() # Get performance metrics @@ -187,7 +192,7 @@ def test_multiple_windows(test_db_session, sample_official, sample_security, sam session = test_db_session """Test calculating returns for multiple windows.""" spy = session.query(Security).filter_by(ticker="SPY").first() - + trade = Trade( official_id=sample_official.id, security_id=spy.id, @@ -231,7 +236,7 @@ def test_sector_analysis(test_db_session, sample_official, sample_prices): value_max=Decimal("50000"), ) session.add(trade) - + session.commit() metrics = PerformanceMetrics(session) @@ -257,7 +262,7 @@ def test_timing_analysis(test_db_session, sample_official, sample_security): value_max=Decimal("50000"), ) session.add(trade) - + session.commit() metrics = PerformanceMetrics(session) @@ -265,4 +270,3 @@ def test_timing_analysis(test_db_session, sample_official, sample_security): assert "avg_disclosure_lag_days" in timing assert timing["avg_disclosure_lag_days"] > 0 - diff --git a/tests/test_analytics_integration.py b/tests/test_analytics_integration.py index d30a026..c443098 100644 --- a/tests/test_analytics_integration.py +++ b/tests/test_analytics_integration.py @@ -1,13 +1,14 @@ """Integration tests for analytics with real-ish data.""" -import pytest from datetime import date, timedelta from decimal import Decimal -from pote.analytics.returns import ReturnCalculator +import pytest + from pote.analytics.benchmarks import BenchmarkComparison from pote.analytics.metrics import PerformanceMetrics -from pote.db.models import Official, Security, Trade, Price +from pote.analytics.returns import ReturnCalculator +from pote.db.models import Official, Price, Security, Trade @pytest.fixture @@ -39,13 +40,13 @@ def full_test_data(test_db_session): # 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, @@ -59,12 +60,12 @@ def full_test_data(test_db_session): # 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, @@ -88,7 +89,7 @@ def full_test_data(test_db_session): value_min=Decimal("15001"), value_max=Decimal("50000"), ) - + # Tuberville buys NVDA later (still good but less alpha) trade2 = Trade( official_id=tuberville.id, @@ -115,22 +116,24 @@ 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']}" - + 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']}") @@ -140,22 +143,22 @@ 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(f"\nāœ… Benchmark Comparison:") + + 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}%") @@ -165,21 +168,21 @@ 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}%") @@ -191,17 +194,17 @@ 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(f"\nāœ… Multiple Windows:") + 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}%") @@ -211,13 +214,13 @@ 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(f"\nāœ… Top Performers:") + + 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}%") @@ -227,18 +230,18 @@ 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(f"\nāœ… System Statistics:") + + 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%}") @@ -248,13 +251,13 @@ 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(f"\nāœ… Disclosure Timing:") + + 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") @@ -263,25 +266,27 @@ 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(f"\nāœ… Sector Analysis:") + print("\nāœ… Sector Analysis:") for s in sectors: - print(f" {s['sector']:15s} | {s['trade_count']} trades | Alpha: {s['avg_alpha']:+6.2f}%") + 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, @@ -294,9 +299,9 @@ def test_edge_case_missing_exit_price(test_db_session, full_test_data): ) 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") @@ -305,10 +310,10 @@ 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, @@ -321,11 +326,10 @@ def test_sell_trade_logic(test_db_session, full_test_data): ) 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}%") - diff --git a/tests/test_disclosure_correlator.py b/tests/test_disclosure_correlator.py index 77edc1d..f7239b0 100644 --- a/tests/test_disclosure_correlator.py +++ b/tests/test_disclosure_correlator.py @@ -1,24 +1,25 @@ """Tests for disclosure correlation module.""" -import pytest -from datetime import date, datetime, timedelta, timezone +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 -from pote.db.models import Official, Security, Trade, MarketAlert @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, @@ -32,13 +33,13 @@ def trade_with_alerts(test_db_session): ) 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=timezone.utc), # 5 days before + timestamp=datetime(2024, 1, 10, 10, 30, tzinfo=UTC), # 5 days before details={"multiplier": 3.5}, price=Decimal("490.00"), volume=100000000, @@ -48,7 +49,7 @@ def trade_with_alerts(test_db_session): MarketAlert( ticker="NVDA", alert_type="price_spike", - timestamp=datetime(2024, 1, 12, 14, 15, tzinfo=timezone.utc), # 3 days before + timestamp=datetime(2024, 1, 12, 14, 15, tzinfo=UTC), # 3 days before details={"change_pct": 5.5}, price=Decimal("505.00"), volume=85000000, @@ -58,7 +59,7 @@ def trade_with_alerts(test_db_session): MarketAlert( ticker="NVDA", alert_type="high_volatility", - timestamp=datetime(2024, 1, 14, 16, 20, tzinfo=timezone.utc), # 1 day before + timestamp=datetime(2024, 1, 14, 16, 20, tzinfo=UTC), # 1 day before details={"multiplier": 2.5}, price=Decimal("510.00"), volume=90000000, @@ -68,7 +69,7 @@ def trade_with_alerts(test_db_session): ] session.add_all(alerts) session.commit() - + return { "trade": trade, "official": pelosi, @@ -81,12 +82,12 @@ def trade_with_alerts(test_db_session): 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, @@ -98,7 +99,7 @@ def trade_without_alerts(test_db_session): ) session.add(trade) session.commit() - + return { "trade": trade, "official": official, @@ -110,12 +111,12 @@ 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) @@ -125,11 +126,11 @@ 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 @@ -137,12 +138,12 @@ def test_calculate_timing_score_high_suspicion(test_db_session, trade_with_alert """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 @@ -155,13 +156,13 @@ 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, @@ -172,9 +173,9 @@ def test_calculate_timing_score_no_alerts(test_db_session): ) 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 @@ -184,13 +185,13 @@ 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, @@ -202,47 +203,47 @@ def test_calculate_timing_score_factors(test_db_session): ) 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=timezone.utc), + 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=timezone.utc), + 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=timezone.utc), # Recent + 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=timezone.utc), # Very recent + 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 @@ -253,11 +254,11 @@ 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" @@ -267,7 +268,7 @@ def test_analyze_trade_full(test_db_session, trade_with_alerts): 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 @@ -281,16 +282,15 @@ def test_analyze_recent_disclosures(test_db_session, trade_with_alerts, trade_wi """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, # Wide window to catch test data - min_timing_score=50 + 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): @@ -301,12 +301,12 @@ 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 @@ -319,13 +319,13 @@ 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"] @@ -334,10 +334,10 @@ 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 @@ -349,9 +349,9 @@ 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"] @@ -361,13 +361,13 @@ 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, @@ -379,29 +379,29 @@ def test_alerts_outside_lookback_window(test_db_session): ) 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=timezone.utc), + 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=timezone.utc), + 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) @@ -410,14 +410,14 @@ 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, @@ -428,28 +428,27 @@ def test_different_ticker_alerts_excluded(test_db_session): ) 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=timezone.utc), + 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=timezone.utc), + 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" - diff --git a/tests/test_models.py b/tests/test_models.py index fad9456..4febdd1 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -5,6 +5,7 @@ Tests for database models. from datetime import date from decimal import Decimal +import pytest from sqlalchemy import select from pote.db.models import Price, Security, Trade @@ -56,12 +57,9 @@ def test_unique_constraints(test_db_session, sample_security): dup_security = Security(ticker="AAPL", name="Apple Duplicate") test_db_session.add(dup_security) - try: + with pytest.raises(IntegrityError): test_db_session.commit() - assert False, "Should have raised IntegrityError" - except IntegrityError: - test_db_session.rollback() - # Expected behavior + test_db_session.rollback() def test_price_unique_per_security_date(test_db_session, sample_security): @@ -83,12 +81,9 @@ def test_price_unique_per_security_date(test_db_session, sample_security): ) test_db_session.add(price2) - try: + with pytest.raises(IntegrityError): test_db_session.commit() - assert False, "Should have raised IntegrityError" - except IntegrityError: - test_db_session.rollback() - # Expected behavior + test_db_session.rollback() def test_trade_queries(test_db_session, sample_official, sample_security): diff --git a/tests/test_monitoring.py b/tests/test_monitoring.py index 0ef5650..ad92cb2 100644 --- a/tests/test_monitoring.py +++ b/tests/test_monitoring.py @@ -1,25 +1,26 @@ """Tests for market monitoring module.""" -import pytest -from datetime import date, datetime, timedelta, timezone +from datetime import UTC, date, datetime, timedelta from decimal import Decimal -from pote.monitoring.market_monitor import MarketMonitor +import pytest + +from pote.db.models import MarketAlert, Official, Security, Trade from pote.monitoring.alert_manager import AlertManager -from pote.db.models import Official, Security, Trade, MarketAlert +from pote.monitoring.market_monitor import MarketMonitor @pytest.fixture def sample_congressional_trades(test_db_session): """Create sample congressional trades for watchlist building.""" 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") msft = Security(ticker="MSFT", name="Microsoft Corporation", sector="Technology") @@ -28,28 +29,58 @@ def sample_congressional_trades(test_db_session): spy = Security(ticker="SPY", name="SPDR S&P 500 ETF", sector="Financial") session.add_all([nvda, msft, aapl, tsla, spy]) session.flush() - + # Create multiple trades (NVDA is most traded) trades = [ - Trade(official_id=pelosi.id, security_id=nvda.id, source="test", - transaction_date=date(2024, 1, 15), side="buy", - value_min=Decimal("15001"), value_max=Decimal("50000")), - Trade(official_id=pelosi.id, security_id=nvda.id, source="test", - transaction_date=date(2024, 2, 1), side="buy", - value_min=Decimal("15001"), value_max=Decimal("50000")), - Trade(official_id=tuberville.id, security_id=nvda.id, source="test", - transaction_date=date(2024, 2, 15), side="buy", - value_min=Decimal("50001"), value_max=Decimal("100000")), - Trade(official_id=pelosi.id, security_id=msft.id, source="test", - transaction_date=date(2024, 1, 20), side="sell", - value_min=Decimal("15001"), value_max=Decimal("50000")), - Trade(official_id=tuberville.id, security_id=aapl.id, source="test", - transaction_date=date(2024, 2, 10), side="buy", - value_min=Decimal("15001"), value_max=Decimal("50000")), + Trade( + official_id=pelosi.id, + security_id=nvda.id, + source="test", + transaction_date=date(2024, 1, 15), + side="buy", + value_min=Decimal("15001"), + value_max=Decimal("50000"), + ), + Trade( + official_id=pelosi.id, + security_id=nvda.id, + source="test", + transaction_date=date(2024, 2, 1), + side="buy", + value_min=Decimal("15001"), + value_max=Decimal("50000"), + ), + Trade( + official_id=tuberville.id, + security_id=nvda.id, + source="test", + transaction_date=date(2024, 2, 15), + side="buy", + value_min=Decimal("50001"), + value_max=Decimal("100000"), + ), + Trade( + official_id=pelosi.id, + security_id=msft.id, + source="test", + transaction_date=date(2024, 1, 20), + side="sell", + value_min=Decimal("15001"), + value_max=Decimal("50000"), + ), + Trade( + official_id=tuberville.id, + security_id=aapl.id, + source="test", + transaction_date=date(2024, 2, 10), + side="buy", + value_min=Decimal("15001"), + value_max=Decimal("50000"), + ), ] session.add_all(trades) session.commit() - + return { "officials": [pelosi, tuberville], "securities": [nvda, msft, aapl, tsla, spy], @@ -61,9 +92,9 @@ def sample_congressional_trades(test_db_session): def sample_alerts(test_db_session): """Create sample market alerts.""" session = test_db_session - - now = datetime.now(timezone.utc) - + + now = datetime.now(UTC) + alerts = [ MarketAlert( ticker="NVDA", @@ -96,10 +127,10 @@ def sample_alerts(test_db_session): severity=5, ), ] - + session.add_all(alerts) session.commit() - + return alerts @@ -107,9 +138,9 @@ def test_get_congressional_watchlist(test_db_session, sample_congressional_trade """Test building watchlist from congressional trades.""" session = test_db_session monitor = MarketMonitor(session) - + watchlist = monitor.get_congressional_watchlist(limit=10) - + assert len(watchlist) > 0 assert "NVDA" in watchlist # Most traded assert watchlist[0] == "NVDA" # Should be first (3 trades) @@ -119,11 +150,11 @@ def test_check_ticker_basic(test_db_session): """Test basic ticker checking (may not find alerts with real data).""" session = test_db_session monitor = MarketMonitor(session) - + # This uses real yfinance data, so alerts depend on current market # We test that it doesn't crash alerts = monitor.check_ticker("AAPL", lookback_days=5) - + assert isinstance(alerts, list) # Each alert should have required fields for alert in alerts: @@ -137,7 +168,7 @@ def test_scan_watchlist_with_mock(test_db_session, sample_congressional_trades, """Test scanning watchlist with mocked data.""" session = test_db_session monitor = MarketMonitor(session) - + # Mock the check_ticker method to return controlled data def mock_check_ticker(ticker, lookback_days=5): if ticker == "NVDA": @@ -145,7 +176,7 @@ def test_scan_watchlist_with_mock(test_db_session, sample_congressional_trades, { "ticker": ticker, "alert_type": "unusual_volume", - "timestamp": datetime.now(timezone.utc), + "timestamp": datetime.now(UTC), "details": {"multiplier": 3.5}, "price": Decimal("500.00"), "volume": 100000000, @@ -154,12 +185,12 @@ def test_scan_watchlist_with_mock(test_db_session, sample_congressional_trades, } ] return [] - + monkeypatch.setattr(monitor, "check_ticker", mock_check_ticker) - + # Scan with limited watchlist alerts = monitor.scan_watchlist(tickers=["NVDA", "MSFT"], lookback_days=5) - + assert len(alerts) == 1 assert alerts[0]["ticker"] == "NVDA" assert alerts[0]["alert_type"] == "unusual_volume" @@ -169,12 +200,12 @@ def test_save_alerts(test_db_session): """Test saving alerts to database.""" session = test_db_session monitor = MarketMonitor(session) - + alerts_data = [ { "ticker": "TSLA", "alert_type": "price_spike", - "timestamp": datetime.now(timezone.utc), + "timestamp": datetime.now(UTC), "details": {"change_pct": 7.5}, "price": Decimal("250.00"), "volume": 75000000, @@ -184,7 +215,7 @@ def test_save_alerts(test_db_session): { "ticker": "TSLA", "alert_type": "unusual_volume", - "timestamp": datetime.now(timezone.utc), + "timestamp": datetime.now(UTC), "details": {"multiplier": 4.0}, "price": Decimal("250.00"), "volume": 120000000, @@ -192,11 +223,11 @@ def test_save_alerts(test_db_session): "severity": 9, }, ] - + saved_count = monitor.save_alerts(alerts_data) - + assert saved_count == 2 - + # Verify in database alerts = session.query(MarketAlert).filter_by(ticker="TSLA").all() assert len(alerts) == 2 @@ -206,21 +237,21 @@ def test_get_recent_alerts(test_db_session, sample_alerts): """Test querying recent alerts.""" session = test_db_session monitor = MarketMonitor(session) - + # Get all alerts all_alerts = monitor.get_recent_alerts(days=1) assert len(all_alerts) >= 3 - + # Filter by ticker nvda_alerts = monitor.get_recent_alerts(ticker="NVDA", days=1) assert len(nvda_alerts) == 2 assert all(a.ticker == "NVDA" for a in nvda_alerts) - + # Filter by alert type volume_alerts = monitor.get_recent_alerts(alert_type="unusual_volume", days=1) assert len(volume_alerts) == 1 assert volume_alerts[0].alert_type == "unusual_volume" - + # Filter by severity high_sev_alerts = monitor.get_recent_alerts(min_severity=6, days=1) assert all(a.severity >= 6 for a in high_sev_alerts) @@ -230,12 +261,12 @@ def test_get_ticker_alert_summary(test_db_session, sample_alerts): """Test alert summary by ticker.""" session = test_db_session monitor = MarketMonitor(session) - + summary = monitor.get_ticker_alert_summary(days=1) - + assert "NVDA" in summary assert "MSFT" in summary - + nvda_summary = summary["NVDA"] assert nvda_summary["alert_count"] == 2 assert nvda_summary["max_severity"] == 7 @@ -246,11 +277,11 @@ def test_alert_manager_format_text(test_db_session, sample_alerts): """Test text formatting of alerts.""" session = test_db_session alert_mgr = AlertManager(session) - + alert = sample_alerts[0] # NVDA unusual volume - + text = alert_mgr.format_alert_text(alert) - + assert "NVDA" in text assert "UNUSUAL VOLUME" in text assert "Severity" in text @@ -261,11 +292,11 @@ def test_alert_manager_format_html(test_db_session, sample_alerts): """Test HTML formatting of alerts.""" session = test_db_session alert_mgr = AlertManager(session) - + alert = sample_alerts[0] - + html = alert_mgr.format_alert_html(alert) - + assert "" in report assert "" in report assert "Market Activity Alerts" in report @@ -333,20 +364,20 @@ def test_alert_manager_empty_alerts(test_db_session): """Test handling empty alert list.""" session = test_db_session alert_mgr = AlertManager(session) - - report = alert_mgr.generate_summary_report([], format="text") - + + report = alert_mgr.generate_summary_report([], output_format="text") + assert "No alerts" in report def test_market_alert_model(test_db_session): """Test MarketAlert model creation and retrieval.""" session = test_db_session - + alert = MarketAlert( ticker="GOOGL", alert_type="price_spike", - timestamp=datetime.now(timezone.utc), + timestamp=datetime.now(UTC), details={"test": "data"}, price=Decimal("140.50"), volume=25000000, @@ -354,13 +385,13 @@ def test_market_alert_model(test_db_session): severity=7, source="test", ) - + session.add(alert) session.commit() - + # Retrieve retrieved = session.query(MarketAlert).filter_by(ticker="GOOGL").first() - + assert retrieved is not None assert retrieved.ticker == "GOOGL" assert retrieved.alert_type == "price_spike" @@ -372,9 +403,9 @@ def test_market_alert_model(test_db_session): def test_alert_timestamp_filtering(test_db_session): """Test filtering alerts by timestamp.""" session = test_db_session - - now = datetime.now(timezone.utc) - + + now = datetime.now(UTC) + # Create alerts at different times old_alert = MarketAlert( ticker="TEST1", @@ -388,20 +419,19 @@ def test_alert_timestamp_filtering(test_db_session): timestamp=now - timedelta(hours=2), severity=5, ) - + session.add_all([old_alert, recent_alert]) session.commit() - + monitor = MarketMonitor(session) - + # Should only get recent alert alerts_1_day = monitor.get_recent_alerts(days=1) test_alerts = [a for a in alerts_1_day if a.ticker.startswith("TEST")] assert len(test_alerts) == 1 assert test_alerts[0].ticker == "TEST2" - + # Should get both with longer lookback alerts_30_days = monitor.get_recent_alerts(days=30) test_alerts = [a for a in alerts_30_days if a.ticker.startswith("TEST")] assert len(test_alerts) == 2 - diff --git a/tests/test_pattern_detector.py b/tests/test_pattern_detector.py index 219baeb..8b209d9 100644 --- a/tests/test_pattern_detector.py +++ b/tests/test_pattern_detector.py @@ -1,38 +1,39 @@ """Tests for pattern detection module.""" -import pytest -from datetime import date, datetime, timedelta, timezone +from datetime import UTC, date, datetime, timedelta from decimal import Decimal +import pytest + +from pote.db.models import MarketAlert, Official, Security, Trade from pote.monitoring.pattern_detector import PatternDetector -from pote.db.models import Official, Security, Trade, MarketAlert @pytest.fixture def multiple_officials_with_patterns(test_db_session): """Create multiple officials with different timing patterns.""" 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") clean_trader = Official(name="Clean Trader", chamber="House", party="Independent", state="TX") - + session.add_all([pelosi, tuberville, clean_trader]) session.flush() - + # Create securities nvda = Security(ticker="NVDA", name="NVIDIA", sector="Technology") msft = Security(ticker="MSFT", name="Microsoft", sector="Technology") xom = Security(ticker="XOM", name="Exxon", sector="Energy") - + session.add_all([nvda, msft, xom]) session.flush() - + # Pelosi - Suspicious pattern (trades with alerts) for i in range(5): - trade_date = date(2024, 1, 15) + timedelta(days=i*30) - + trade_date = date(2024, 1, 15) + timedelta(days=i * 30) + # Create trade trade = Trade( official_id=pelosi.id, @@ -45,24 +46,23 @@ def multiple_officials_with_patterns(test_db_session): ) session.add(trade) session.flush() - + # Create alerts BEFORE trade (suspicious) for j in range(2): alert = MarketAlert( ticker="NVDA", alert_type="unusual_volume", timestamp=datetime.combine( - trade_date - timedelta(days=3+j), - datetime.min.time() - ).replace(tzinfo=timezone.utc), + trade_date - timedelta(days=3 + j), datetime.min.time() + ).replace(tzinfo=UTC), severity=7 + j, ) session.add(alert) - + # Tuberville - Mixed pattern for i in range(4): - trade_date = date(2024, 2, 1) + timedelta(days=i*30) - + trade_date = date(2024, 2, 1) + timedelta(days=i * 30) + trade = Trade( official_id=tuberville.id, security_id=msft.id, @@ -74,24 +74,23 @@ def multiple_officials_with_patterns(test_db_session): ) session.add(trade) session.flush() - + # Only first 2 trades have alerts if i < 2: alert = MarketAlert( ticker="MSFT", alert_type="price_spike", timestamp=datetime.combine( - trade_date - timedelta(days=5), - datetime.min.time() - ).replace(tzinfo=timezone.utc), + trade_date - timedelta(days=5), datetime.min.time() + ).replace(tzinfo=UTC), severity=6, ) session.add(alert) - + # Clean trader - No suspicious activity for i in range(3): - trade_date = date(2024, 3, 1) + timedelta(days=i*30) - + trade_date = date(2024, 3, 1) + timedelta(days=i * 30) + trade = Trade( official_id=clean_trader.id, security_id=xom.id, @@ -102,9 +101,9 @@ def multiple_officials_with_patterns(test_db_session): value_max=Decimal("50000"), ) session.add(trade) - + session.commit() - + return { "officials": [pelosi, tuberville, clean_trader], "securities": [nvda, msft, xom], @@ -115,15 +114,15 @@ def test_rank_officials_by_timing(test_db_session, multiple_officials_with_patte """Test ranking officials by timing scores.""" session = test_db_session detector = PatternDetector(session) - + rankings = detector.rank_officials_by_timing(lookback_days=3650, min_trades=3) - + assert len(rankings) >= 2 # At least 2 officials with 3+ trades - + # Rankings should be sorted by avg_timing_score (descending) for i in range(len(rankings) - 1): assert rankings[i]["avg_timing_score"] >= rankings[i + 1]["avg_timing_score"] - + # Check required fields for ranking in rankings: assert "name" in ranking @@ -138,16 +137,15 @@ def test_identify_repeat_offenders(test_db_session, multiple_officials_with_patt """Test identifying repeat offenders.""" session = test_db_session detector = PatternDetector(session) - + # Set low threshold to catch Pelosi (who has 100% suspicious rate) offenders = detector.identify_repeat_offenders( - lookback_days=3650, - min_suspicious_rate=0.7 # 70%+ + lookback_days=3650, min_suspicious_rate=0.7 # 70%+ ) - + # Should find at least Pelosi (all trades with alerts) assert isinstance(offenders, list) - + # All offenders should have high suspicious rates for offender in offenders: assert offender["suspicious_rate"] >= 70 @@ -157,19 +155,16 @@ def test_analyze_ticker_patterns(test_db_session, multiple_officials_with_patter """Test ticker pattern analysis.""" session = test_db_session detector = PatternDetector(session) - - ticker_patterns = detector.analyze_ticker_patterns( - lookback_days=3650, - min_trades=3 - ) - + + ticker_patterns = detector.analyze_ticker_patterns(lookback_days=3650, min_trades=3) + assert isinstance(ticker_patterns, list) assert len(ticker_patterns) >= 1 # At least NVDA should qualify - + # Check sorting for i in range(len(ticker_patterns) - 1): assert ticker_patterns[i]["avg_timing_score"] >= ticker_patterns[i + 1]["avg_timing_score"] - + # Check fields for pattern in ticker_patterns: assert "ticker" in pattern @@ -182,12 +177,12 @@ def test_get_sector_timing_analysis(test_db_session, multiple_officials_with_pat """Test sector timing analysis.""" session = test_db_session detector = PatternDetector(session) - + sector_stats = detector.get_sector_timing_analysis(lookback_days=3650) - + assert isinstance(sector_stats, dict) assert len(sector_stats) >= 2 # Technology and Energy - + # Check Technology sector (should have alerts) if "Technology" in sector_stats: tech = sector_stats["Technology"] @@ -201,14 +196,14 @@ def test_get_party_comparison(test_db_session, multiple_officials_with_patterns) """Test party comparison analysis.""" session = test_db_session detector = PatternDetector(session) - + party_stats = detector.get_party_comparison(lookback_days=3650) - + assert isinstance(party_stats, dict) assert len(party_stats) >= 2 # Democrat, Republican, Independent - + # Check that we have data for each party - for party, stats in party_stats.items(): + for stats in party_stats.values(): assert "official_count" in stats assert "total_trades" in stats assert "avg_timing_score" in stats @@ -219,9 +214,9 @@ def test_generate_pattern_report(test_db_session, multiple_officials_with_patter """Test comprehensive pattern report generation.""" session = test_db_session detector = PatternDetector(session) - + report = detector.generate_pattern_report(lookback_days=3650) - + # Check report structure assert "period_days" in report assert "summary" in report @@ -230,12 +225,12 @@ def test_generate_pattern_report(test_db_session, multiple_officials_with_patter assert "suspicious_tickers" in report assert "sector_analysis" in report assert "party_comparison" in report - + # Check summary summary = report["summary"] assert summary["total_officials_analyzed"] >= 2 assert "avg_timing_score" in summary - + # Check that lists are populated assert len(report["top_suspicious_officials"]) >= 2 assert isinstance(report["suspicious_tickers"], list) @@ -245,15 +240,15 @@ def test_rank_officials_min_trades_filter(test_db_session, multiple_officials_wi """Test that min_trades filter works correctly.""" session = test_db_session detector = PatternDetector(session) - + # With min_trades=5, should only get Pelosi rankings_high = detector.rank_officials_by_timing(lookback_days=3650, min_trades=5) - + # With min_trades=3, should get at least 2 officials rankings_low = detector.rank_officials_by_timing(lookback_days=3650, min_trades=3) - + assert len(rankings_low) >= len(rankings_high) - + # All officials should meet min_trades requirement for ranking in rankings_high: assert ranking["trade_count"] >= 5 @@ -263,17 +258,17 @@ def test_empty_data_handling(test_db_session): """Test handling of empty dataset.""" session = test_db_session detector = PatternDetector(session) - + # With no data, should return empty results rankings = detector.rank_officials_by_timing(lookback_days=30, min_trades=1) assert rankings == [] - + offenders = detector.identify_repeat_offenders(lookback_days=30) assert offenders == [] - + tickers = detector.analyze_ticker_patterns(lookback_days=30) assert tickers == [] - + sectors = detector.get_sector_timing_analysis(lookback_days=30) assert sectors == {} @@ -282,13 +277,13 @@ def test_ranking_score_accuracy(test_db_session, multiple_officials_with_pattern """Test that rankings accurately reflect timing patterns.""" session = test_db_session detector = PatternDetector(session) - + rankings = detector.rank_officials_by_timing(lookback_days=3650, min_trades=3) - + # Find Pelosi and Clean Trader pelosi_rank = next((r for r in rankings if "Pelosi" in r["name"]), None) clean_rank = next((r for r in rankings if "Clean" in r["name"]), None) - + if pelosi_rank and clean_rank: # Pelosi (with alerts) should have higher score than clean trader (no alerts) assert pelosi_rank["avg_timing_score"] > clean_rank["avg_timing_score"] @@ -299,9 +294,9 @@ def test_sector_stats_accuracy(test_db_session, multiple_officials_with_patterns """Test sector statistics are calculated correctly.""" session = test_db_session detector = PatternDetector(session) - + sector_stats = detector.get_sector_timing_analysis(lookback_days=3650) - + # Energy should have clean pattern (no alerts) if "Energy" in sector_stats: energy = sector_stats["Energy"] @@ -313,14 +308,12 @@ def test_party_stats_completeness(test_db_session, multiple_officials_with_patte """Test party statistics completeness.""" session = test_db_session detector = PatternDetector(session) - + party_stats = detector.get_party_comparison(lookback_days=3650) - + # Check Democrats (Pelosi) if "Democrat" in party_stats: dem = party_stats["Democrat"] assert dem["official_count"] >= 1 assert dem["total_trades"] >= 5 # Pelosi has 5 trades assert dem["total_suspicious"] > 0 # Pelosi has suspicious trades - -
TickerTrade Count