You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
192 lines
7.0 KiB
Python
192 lines
7.0 KiB
Python
"""
|
|
PIT (point-in-time) correctness + business logic tests for Form 4.
|
|
|
|
Pure unit tests (no DB) for c-suite derivation, purchase_pct formula,
|
|
and service query-building logic.
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import date, datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from app.services.insider_transaction_service import InsiderTransactionService
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# C-suite title derivation
|
|
# ------------------------------------------------------------------
|
|
|
|
class TestDeriveTitleFlags:
|
|
def test_ceo_full_title(self):
|
|
flags = InsiderTransactionService._derive_title_flags("Chief Executive Officer")
|
|
assert flags == {"is_ceo": True, "is_cfo": False, "is_c_suite": True}
|
|
|
|
def test_ceo_abbrev(self):
|
|
flags = InsiderTransactionService._derive_title_flags("CEO")
|
|
assert flags["is_ceo"] is True
|
|
assert flags["is_c_suite"] is True
|
|
|
|
def test_cfo_full_title(self):
|
|
flags = InsiderTransactionService._derive_title_flags("Chief Financial Officer")
|
|
assert flags["is_cfo"] is True
|
|
assert flags["is_c_suite"] is True
|
|
|
|
def test_principal_financial_officer(self):
|
|
flags = InsiderTransactionService._derive_title_flags("Principal Financial Officer")
|
|
assert flags["is_cfo"] is True
|
|
|
|
def test_president(self):
|
|
flags = InsiderTransactionService._derive_title_flags("President, Americas Division")
|
|
assert flags["is_c_suite"] is True
|
|
assert flags["is_ceo"] is False
|
|
|
|
def test_chairman(self):
|
|
flags = InsiderTransactionService._derive_title_flags("Chairman of the Board")
|
|
assert flags["is_c_suite"] is True
|
|
|
|
def test_coo(self):
|
|
flags = InsiderTransactionService._derive_title_flags("COO")
|
|
assert flags["is_c_suite"] is True
|
|
|
|
def test_evp_is_not_csuite(self):
|
|
# "EVP Sales" does not match CEO/CFO/COO/... or "Chief X Officer"
|
|
flags = InsiderTransactionService._derive_title_flags("EVP Sales")
|
|
assert flags["is_c_suite"] is False
|
|
|
|
def test_director_is_not_csuite(self):
|
|
flags = InsiderTransactionService._derive_title_flags("Director")
|
|
assert all(not v for v in flags.values())
|
|
|
|
def test_none_title(self):
|
|
flags = InsiderTransactionService._derive_title_flags(None)
|
|
assert flags == {"is_ceo": False, "is_cfo": False, "is_c_suite": False}
|
|
|
|
def test_empty_string(self):
|
|
flags = InsiderTransactionService._derive_title_flags("")
|
|
assert all(not v for v in flags.values())
|
|
|
|
def test_case_insensitive(self):
|
|
flags = InsiderTransactionService._derive_title_flags("ceo")
|
|
assert flags["is_ceo"] is True
|
|
|
|
def test_ceo_and_president(self):
|
|
flags = InsiderTransactionService._derive_title_flags("President and CEO")
|
|
assert flags["is_ceo"] is True
|
|
assert flags["is_c_suite"] is True
|
|
|
|
def test_chief_marketing_officer(self):
|
|
flags = InsiderTransactionService._derive_title_flags("Chief Marketing Officer")
|
|
assert flags["is_c_suite"] is True
|
|
assert flags["is_ceo"] is False
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# purchase_pct_of_holding formula (tested via _parse_transaction_element mock)
|
|
# ------------------------------------------------------------------
|
|
|
|
class TestPurchasePct:
|
|
"""Test the purchase_pct_of_holding calculation logic directly."""
|
|
|
|
def _compute_pct(self, code, shares, shares_after):
|
|
"""Mirrors the formula in _parse_transaction_element."""
|
|
if code in ("P", "A") and shares is not None and shares > 0 and shares_after and shares_after > 0:
|
|
return abs(shares) / shares_after
|
|
return None
|
|
|
|
def test_open_market_buy(self):
|
|
pct = self._compute_pct("P", 100.0, 1000.0)
|
|
assert pct == pytest.approx(0.1, rel=1e-6)
|
|
|
|
def test_award(self):
|
|
pct = self._compute_pct("A", 500.0, 5000.0)
|
|
assert pct == pytest.approx(0.1, rel=1e-6)
|
|
|
|
def test_sell_is_none(self):
|
|
assert self._compute_pct("S", -100.0, 900.0) is None
|
|
|
|
def test_zero_shares_after(self):
|
|
assert self._compute_pct("P", 100.0, 0.0) is None
|
|
|
|
def test_none_shares_after(self):
|
|
assert self._compute_pct("P", 100.0, None) is None
|
|
|
|
def test_negative_buy_shares_is_none(self):
|
|
# Negative shares on a "P" code shouldn't happen, but guard
|
|
assert self._compute_pct("P", -100.0, 1000.0) is None
|
|
|
|
def test_small_buy(self):
|
|
pct = self._compute_pct("P", 1.0, 1_000_000.0)
|
|
assert pct == pytest.approx(1e-6, rel=1e-4)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# PIT filter logic (mocked DB)
|
|
# ------------------------------------------------------------------
|
|
|
|
class TestGetForm4Pit:
|
|
"""Verify that get_form4_pit builds correct date conditions."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_as_of_is_used_as_upper_bound(self):
|
|
svc = InsiderTransactionService()
|
|
mock_db = AsyncMock()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar.return_value = 0
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
mock_db.execute = AsyncMock(return_value=mock_result)
|
|
|
|
rows, total = await svc.get_form4_pit(
|
|
mock_db, ticker="AAPL", as_of=date(2024, 1, 15)
|
|
)
|
|
assert total == 0
|
|
assert rows == []
|
|
# Verify execute was called (conditions were built)
|
|
assert mock_db.execute.called
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_buy_only_adds_code_filter(self):
|
|
svc = InsiderTransactionService()
|
|
mock_db = AsyncMock()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar.return_value = 0
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
mock_db.execute = AsyncMock(return_value=mock_result)
|
|
|
|
# Should not raise
|
|
rows, total = await svc.get_form4_pit(
|
|
mock_db, ticker="NVDA", as_of=date(2024, 3, 5), buy_only=True
|
|
)
|
|
assert mock_db.execute.called
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Aggregate query
|
|
# ------------------------------------------------------------------
|
|
|
|
class TestGetForm4Aggregate:
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_dict_structure(self):
|
|
svc = InsiderTransactionService()
|
|
mock_db = AsyncMock()
|
|
mock_row = MagicMock()
|
|
mock_row.buy_count = 3
|
|
mock_row.buy_dollar_total = 150000.0
|
|
mock_row.cluster_size = 2
|
|
mock_row.csuite_count = 1
|
|
mock_row.avg_pct_of_holding = 0.05
|
|
mock_row.last_filing_date = datetime(2024, 1, 10, tzinfo=timezone.utc)
|
|
mock_result = MagicMock()
|
|
mock_result.one.return_value = mock_row
|
|
mock_db.execute = AsyncMock(return_value=mock_result)
|
|
|
|
agg = await svc.get_form4_aggregate(
|
|
mock_db, ticker="AAPL", as_of=date(2024, 1, 15), window_days=30
|
|
)
|
|
assert agg["symbol"] == "AAPL"
|
|
assert agg["buy_count"] == 3
|
|
assert agg["cluster_size"] == 2
|
|
assert agg["csuite_count"] == 1
|
|
assert "recency_days" in agg
|
|
assert "buy_dollar_total" in agg
|