invest-copilot/tasks/phase2-wire-financials-ingestion/SPEC.md

Task: Wire Financials Ingestion as Celery Task

Status: COMPLETE

Current State

src/data-pipeline/ingest_financials.py exists as a standalone script. It uses psycopg2 (sync) and runs as a CLI script.

Goal

Make financial data ingestion runnable as a scheduled Celery task, integrated into the pipeline orchestrator.

Requirements

1. Create src/backend/tasks/ingest_financials.py Celery task
2. Reuse logic from src/data-pipeline/ingest_financials.py (don't duplicate)
3. Task must be idempotent — deduplicate by (ticker, report_type, period)
4. Task must store results into PostgreSQL/TimescaleDB using existing models
5. Add to Celery Beat schedule in celery_app.py
6. Add to docker-compose worker service

Acceptance Criteria

• celery -A celery_app worker processes financial ingestion tasks
• Running the task twice with same input produces no duplicate records (ON CONFLICT)
• Financial data visible in database after task completes
• Celery Beat runs it on a configurable schedule (default: daily at 2 AM UTC)

Implementation Details

Files Created

• src/backend/tasks/ingest_financials.py — Celery task using yfinance + asyncpg
  • Fetches quarterly + annual income statement, balance sheet, cash flow
  • Extracts 20 metrics (revenue, net_income, eps, roe, roa, etc.)
  • Idempotent upsert via ON CONFLICT (ticker, filing_date, period)
  • 3 retries with exponential backoff for both yfinance fetches and DB writes

Files Modified

• src/backend/tasks/__init__.py — Added ingest_financials_task export
• src/backend/celery_app.py — Added ingest-financials-daily schedule (2 AM UTC)
• init.sql — Added financials table + indexes (was missing from Docker init)
• src/backend/database.py — Added financials table to init_db() (was missing)

Key Design Decisions

• Used yfinance (already in backend requirements.txt) instead of duplicating the data-pipeline script
• Converted from psycopg2 to asyncpg via _execute_command() wrapper
• Extracted metrics via column-name matching (yfinance column names vary by ticker)
• Derived metrics: debt_to_equity, roe, roa computed from base values

Next Steps After This Task

Wire news ingestion (phase2-wire-news-ingestion)