Database Operations

Golden Rule

Data goes in the database. No random CSVs/JSON outside data/ directory.

• JSON/CSV = raw, unclean, not for production
• PostgreSQL = production data storage

DBManager Interface

from db_manager import DBManager, default_db

# Use default connection
df = default_db.fetch_df("SELECT * FROM unified_games WHERE sport = 'nba'")

# Or create custom connection
db = DBManager(connection_string="postgresql://user:pass@host:5432/db")

Key Tables

Table	Purpose
unified_games	All games across sports
game_odds	Kalshi/sportsbook odds
placed_bets	Bet history and results
elo_ratings	Historical Elo snapshots
portfolio_snapshots	Portfolio value over time

Common Operations

Fetch as DataFrame

df = default_db.fetch_df("""
    SELECT * FROM unified_games
    WHERE sport = :sport AND game_date >= :start_date
""", {"sport": "nba", "start_date": "2024-01-01"})

Execute Query

default_db.execute("""
    UPDATE placed_bets SET status = 'won'
    WHERE bet_id = :bet_id
""", {"bet_id": "123"})

Insert DataFrame

df.to_sql("unified_games", default_db.get_engine(),
          if_exists="append", index=False)

Upsert Pattern

default_db.execute("""
    INSERT INTO game_odds (game_id, source, yes_price)
    VALUES (:game_id, :source, :yes_price)
    ON CONFLICT (game_id, source)
    DO UPDATE SET yes_price = EXCLUDED.yes_price
""", params)

Connection Defaults

From environment or docker-compose:

• Host: localhost
• Port: 5432
• User: airflow
• Password: airflow
• Database: airflow

Data Validation

Always validate before commits:

from data_validation import validate_nba_data

report = validate_nba_data()
if not report.is_valid:
    raise ValueError(f"Validation failed: {report.errors}")

Files to Reference

• plugins/db_manager.py - DBManager class
• plugins/data_validation.py - Validation utilities
• plugins/database_schema_manager.py - Schema definitions