diff --git a/apps/pipeline/feature_builder/main.py b/apps/pipeline/feature_builder/main.py index d64870f..30fbbb5 100644 --- a/apps/pipeline/feature_builder/main.py +++ b/apps/pipeline/feature_builder/main.py @@ -15,6 +15,7 @@ from libs.db.models import Event, JobRun from libs.db.session import get_session from libs.features.builder import build_features_for_event from libs.oracle_client.client import make_oracle_client +from libs.oracle_client.financial import FinancialService from libs.oracle_client.price import PriceService logger = get_logger(__name__) @@ -25,6 +26,7 @@ async def run_feature_builder(run_id: str) -> dict[str, int]: async with make_oracle_client() as client: price_svc = PriceService(client) + financial_svc = FinancialService(client) async with get_session() as session: job = JobRun( @@ -45,7 +47,9 @@ async def run_feature_builder(run_id: str) -> dict[str, int]: for event in events: try: - snapshots = await build_features_for_event(session, event, price_svc) + snapshots = await build_features_for_event( + session, event, price_svc, financial_service=financial_svc + ) if snapshots is None: event.status = "rejected" event.updated_at_utc = dt.datetime.now(tz=dt.UTC) diff --git a/apps/pipeline/filing_poller/main.py b/apps/pipeline/filing_poller/main.py index b64cf71..003a75e 100644 --- a/apps/pipeline/filing_poller/main.py +++ b/apps/pipeline/filing_poller/main.py @@ -20,12 +20,18 @@ from libs.oracle_client.filings import FilingsService logger = get_logger(__name__) -async def poll_filings(run_id: str) -> dict[str, int]: +async def poll_filings( + run_id: str, + start_date: str | None = None, + end_date: str | None = None, +) -> dict[str, int]: settings = get_settings() symbols = settings.get_symbols() app_config = settings.get_app_config() form_types = ",".join(app_config.get("pipeline", {}).get("form_types", ["8-K", "6-K"])) + effective_start = start_date or (dt.date.today() - dt.timedelta(days=7)).isoformat() + stats = {"seen": 0, "written": 0, "skipped": 0, "errors": 0} async with make_oracle_client() as client: @@ -48,7 +54,8 @@ async def poll_filings(run_id: str) -> dict[str, int]: response = await svc.search_filings( ticker, form_type=form_types, - start_date=(dt.date.today() - dt.timedelta(days=7)).isoformat(), + start_date=effective_start, + end_date=end_date, ) stats["seen"] += len(response.filings) @@ -116,13 +123,25 @@ async def poll_filings(run_id: str) -> dict[str, int]: def main() -> None: parser = argparse.ArgumentParser(description="Filing Poller") parser.add_argument("--run-id", default=new_job_run_id()) + parser.add_argument( + "--start-date", + default=None, + metavar="YYYY-MM-DD", + help="Start date for filing search (default: 7 days ago)", + ) + parser.add_argument( + "--end-date", + default=None, + metavar="YYYY-MM-DD", + help="End date for filing search (default: today)", + ) args = parser.parse_args() settings = get_settings() configure_logging(settings.log_level) bind_job_run_id(args.run_id) - asyncio.run(poll_filings(args.run_id)) + asyncio.run(poll_filings(args.run_id, start_date=args.start_date, end_date=args.end_date)) if __name__ == "__main__": diff --git a/libs/features/builder.py b/libs/features/builder.py index 9743964..c86475c 100644 --- a/libs/features/builder.py +++ b/libs/features/builder.py @@ -1,4 +1,4 @@ -"""Feature orchestrator: combines market and event features.""" +"""Feature orchestrator: combines market, event, and financial features.""" from __future__ import annotations import datetime as dt @@ -9,7 +9,9 @@ from sqlalchemy.ext.asyncio import AsyncSession from libs.common.logging import get_logger from libs.db.models import Event, EventParse, FeatureSnapshot from libs.features.event_features import compute_event_features +from libs.features.financial_features import compute_financial_features from libs.features.market_features import compute_market_features +from libs.oracle_client.financial import FinancialService from libs.oracle_client.price import PriceService logger = get_logger(__name__) @@ -21,6 +23,7 @@ async def build_features_for_event( session: AsyncSession, event: Event, price_service: PriceService, + financial_service: FinancialService | None = None, ) -> tuple[FeatureSnapshot, FeatureSnapshot] | None: """Build market_v1 and event_v1 feature snapshots for an event. @@ -89,6 +92,29 @@ async def build_features_for_event( session.add(event_snapshot) await session.flush() + # Optional: financial features (non-fatal if unavailable) + if financial_service is not None: + try: + fin_response = await financial_service.get_financial_data(ticker) + ff = compute_financial_features(fin_response) + if ff: + financial_snapshot = FeatureSnapshot( + event_id=event.event_id, + snapshot_name="financial_v1", + snapshot_version=SNAPSHOT_VERSION, + feature_json=ff, + ) + session.add(financial_snapshot) + await session.flush() + logger.info("financial_features_built", event_id=event.event_id, ticker=ticker) + except Exception as exc: + logger.warning( + "financial_features_skipped", + event_id=event.event_id, + ticker=ticker, + error=str(exc), + ) + logger.info( "features_built", event_id=event.event_id, diff --git a/libs/features/financial_features.py b/libs/features/financial_features.py new file mode 100644 index 0000000..fbca281 --- /dev/null +++ b/libs/features/financial_features.py @@ -0,0 +1,41 @@ +"""Financial feature calculations from Oracle financial data.""" +from __future__ import annotations + +from typing import Any + +from libs.oracle_client.models import FinancialDataResponse + + +def compute_financial_features(response: FinancialDataResponse) -> dict[str, Any]: + """Compute financial features from quarterly period data. + + Returns dict with latest-quarter metrics and QoQ growth rates. + Returns empty dict if no period data is available. + """ + periods = sorted(response.periods, key=lambda p: p.period_end, reverse=True) + if not periods: + return {} + + latest = periods[0] + features: dict[str, Any] = { + "latest_eps": latest.eps, + "latest_gross_margin": latest.gross_margin, + "latest_operating_margin": latest.operating_margin, + "eps_growth_qoq": None, + "revenue_growth_qoq": None, + } + + if len(periods) >= 2: + prior = periods[1] + if latest.eps is not None and prior.eps is not None and prior.eps != 0: + features["eps_growth_qoq"] = (latest.eps - prior.eps) / abs(prior.eps) + if ( + latest.revenue is not None + and prior.revenue is not None + and prior.revenue != 0 + ): + features["revenue_growth_qoq"] = ( + latest.revenue - prior.revenue + ) / prior.revenue + + return features diff --git a/libs/oracle_client/client.py b/libs/oracle_client/client.py index 03a8fab..d085945 100644 --- a/libs/oracle_client/client.py +++ b/libs/oracle_client/client.py @@ -5,6 +5,7 @@ from typing import Any import httpx +from libs.common.retries import with_retry from libs.oracle_client.exceptions import ( OracleClientError, OracleConnectionError, @@ -39,6 +40,7 @@ class OracleClient: raise RuntimeError("OracleClient must be used as async context manager.") return self._client + @with_retry(max_attempts=3, min_wait=0.1, max_wait=5.0, multiplier=0.1) async def get(self, path: str, params: dict[str, Any] | None = None) -> Any: client = self._ensure_client() try: @@ -54,6 +56,7 @@ class OracleClient: return self._handle_response(response, path) + @with_retry(max_attempts=3, min_wait=0.1, max_wait=5.0, multiplier=0.1) async def post(self, path: str, json: dict[str, Any] | None = None) -> Any: client = self._ensure_client() try: diff --git a/tests/unit/test_financial_features.py b/tests/unit/test_financial_features.py new file mode 100644 index 0000000..8da044e --- /dev/null +++ b/tests/unit/test_financial_features.py @@ -0,0 +1,110 @@ +"""Unit tests for financial feature calculations.""" +import pytest + +from libs.oracle_client.models import FinancialDataResponse, FinancialPeriod + +TWO_PERIOD_RESPONSE = FinancialDataResponse( + ticker="AAPL", + periods=[ + FinancialPeriod( + period="2026-Q1", + period_end="2025-12-28", + revenue=124_300_000_000, + net_income=36_000_000_000, + eps=2.34, + gross_margin=0.472, + operating_margin=0.315, + ), + FinancialPeriod( + period="2025-Q4", + period_end="2025-09-27", + revenue=119_600_000_000, + net_income=34_900_000_000, + eps=2.26, + gross_margin=0.461, + operating_margin=0.308, + ), + ], +) + + +def test_compute_financial_features_latest_values(): + from libs.features.financial_features import compute_financial_features + + features = compute_financial_features(TWO_PERIOD_RESPONSE) + + assert features["latest_eps"] == pytest.approx(2.34) + assert features["latest_gross_margin"] == pytest.approx(0.472) + assert features["latest_operating_margin"] == pytest.approx(0.315) + + +def test_eps_growth_qoq(): + from libs.features.financial_features import compute_financial_features + + features = compute_financial_features(TWO_PERIOD_RESPONSE) + + expected = (2.34 - 2.26) / abs(2.26) + assert features["eps_growth_qoq"] == pytest.approx(expected) + + +def test_revenue_growth_qoq(): + from libs.features.financial_features import compute_financial_features + + features = compute_financial_features(TWO_PERIOD_RESPONSE) + + expected = (124_300_000_000 - 119_600_000_000) / 119_600_000_000 + assert features["revenue_growth_qoq"] == pytest.approx(expected) + + +def test_periods_sorted_by_period_end_descending(): + from libs.features.financial_features import compute_financial_features + + # Provide periods out of chronological order; latest should still be picked + response = FinancialDataResponse( + ticker="AAPL", + periods=[ + FinancialPeriod(period="2025-Q4", period_end="2025-09-27", eps=2.26), + FinancialPeriod(period="2026-Q1", period_end="2025-12-28", eps=2.34), + ], + ) + features = compute_financial_features(response) + assert features["latest_eps"] == pytest.approx(2.34) + + +def test_single_period_no_growth_fields(): + from libs.features.financial_features import compute_financial_features + + response = FinancialDataResponse( + ticker="AAPL", + periods=[ + FinancialPeriod( + period="2026-Q1", period_end="2025-12-28", eps=2.34, gross_margin=0.472 + ) + ], + ) + features = compute_financial_features(response) + + assert features["latest_eps"] == pytest.approx(2.34) + assert features["eps_growth_qoq"] is None + assert features["revenue_growth_qoq"] is None + + +def test_empty_periods_returns_empty_dict(): + from libs.features.financial_features import compute_financial_features + + response = FinancialDataResponse(ticker="AAPL", periods=[]) + assert compute_financial_features(response) == {} + + +def test_none_eps_in_prior_skips_growth(): + from libs.features.financial_features import compute_financial_features + + response = FinancialDataResponse( + ticker="AAPL", + periods=[ + FinancialPeriod(period="2026-Q1", period_end="2025-12-28", eps=2.34), + FinancialPeriod(period="2025-Q4", period_end="2025-09-27", eps=None), + ], + ) + features = compute_financial_features(response) + assert features["eps_growth_qoq"] is None diff --git a/tests/unit/test_oracle_client.py b/tests/unit/test_oracle_client.py index 9062dd8..820e35b 100644 --- a/tests/unit/test_oracle_client.py +++ b/tests/unit/test_oracle_client.py @@ -82,6 +82,9 @@ async def test_server_error_raises_oracle_server_error(httpx_mock: HTTPXMock): from libs.oracle_client.exceptions import OracleServerError from libs.oracle_client.filings import FilingsService + # Must provide one response per retry attempt (3 total) + httpx_mock.add_response(status_code=500) + httpx_mock.add_response(status_code=500) httpx_mock.add_response(status_code=500) async with OracleClient("http://oracle:18001") as client: @@ -90,6 +93,23 @@ async def test_server_error_raises_oracle_server_error(httpx_mock: HTTPXMock): await svc.get_exhibit("ACC123") +@pytest.mark.asyncio +async def test_get_retries_on_transient_error_then_succeeds(httpx_mock: HTTPXMock): + from libs.oracle_client.client import OracleClient + from libs.oracle_client.filings import FilingsService + + data = load_fixture("exhibit_content.json") + httpx_mock.add_response(status_code=500) # attempt 1 fails + httpx_mock.add_response(status_code=500) # attempt 2 fails + httpx_mock.add_response(json=data) # attempt 3 succeeds + + async with OracleClient("http://oracle:18001") as client: + svc = FilingsService(client) + result = await svc.get_exhibit("0000320193-26-000001") + + assert result.accession_no == "0000320193-26-000001" + + @pytest.mark.asyncio async def test_get_short_volume(httpx_mock: HTTPXMock): from libs.oracle_client.client import OracleClient