Database Models and Alembic Migrations in the AI Hedge Fund Web Backend

The AI Hedge Fund backend uses SQLAlchemy models (HedgeFundFlow, HedgeFundFlowRun, HedgeFundFlowRunCycle, ApiKey) stored in SQLite and versioned through defensive Alembic migrations to persist React Flow configurations, execution runs, and trading analytics.

The virattt/ai-hedge-fund project implements a sophisticated web backend that relies on structured database models and Alembic migrations to manage persistent state for algorithmic trading workflows. This architecture stores React Flow configurations, execution histories, and API credentials in SQLite tables defined by SQLAlchemy ORM classes, with schema changes tracked through incremental migration scripts located in app/backend/alembic/versions/.

Core SQLAlchemy Models in the Backend

All database models inherit from a shared declarative base defined in app/backend/database/connection.py and are implemented in app/backend/database/models.py. These models map directly to SQLite tables that store the application's core entities.

Flow Configuration Storage (HedgeFundFlow)

The HedgeFundFlow class persists React Flow visual configurations including nodes, edges, and viewport states for trading strategy workflows.

class HedgeFundFlow(Base):
    __tablename__ = "hedge_fund_flows"

    id = Column(Integer, primary_key=True, index=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
    name = Column(String(200), nullable=False)
    description = Column(Text, nullable=True)
    nodes = Column(JSON, nullable=False)          # React Flow nodes

    edges = Column(JSON, nullable=False)          # React Flow edges

    viewport = Column(JSON, nullable=True)        # Zoom / pan state

    data = Column(JSON, nullable=True)            # Node-internal state

    is_template = Column(Boolean, default=False)
    tags = Column(JSON, nullable=True)

Execution Tracking Models

The backend tracks algorithmic trading executions through two related models. HedgeFundFlowRun records high-level execution metadata, while HedgeFundFlowRunCycle captures granular data from individual analysis-trading cycles.

HedgeFundFlowRun (hedge_fund_flow_runs table):

class HedgeFundFlowRun(Base):
    __tablename__ = "hedge_fund_flow_runs"

    id = Column(Integer, primary_key=True, index=True)
    flow_id = Column(Integer, ForeignKey("hedge_fund_flows.id"), nullable=False, index=True)
    status = Column(String(50), nullable=False, default="IDLE")
    started_at = Column(DateTime(timezone=True), nullable=True)
    completed_at = Column(DateTime(timezone=True), nullable=True)
    trading_mode = Column(String(50), nullable=False, default="one-time")
    schedule = Column(String(50), nullable=True)
    request_data = Column(JSON, nullable=True)
    initial_portfolio = Column(JSON, nullable=True)
    final_portfolio = Column(JSON, nullable=True)
    results = Column(JSON, nullable=True)
    error_message = Column(Text, nullable=True)
    run_number = Column(Integer, nullable=False, default=1)

HedgeFundFlowRunCycle (hedge_fund_flow_run_cycles table):

class HedgeFundFlowRunCycle(Base):
    __tablename__ = "hedge_fund_flow_run_cycles"

    id = Column(Integer, primary_key=True, index=True)
    flow_run_id = Column(Integer, ForeignKey("hedge_fund_flow_runs.id"), nullable=False, index=True)
    cycle_number = Column(Integer, nullable=False)
    started_at = Column(DateTime(timezone=True), nullable=False)
    analyst_signals = Column(JSON, nullable=True)
    trading_decisions = Column(JSON, nullable=True)
    executed_trades = Column(JSON, nullable=True)
    portfolio_snapshot = Column(JSON, nullable=True)
    performance_metrics = Column(JSON, nullable=True)
    status = Column(String(50), nullable=False, default="IN_PROGRESS")
    llm_calls_count = Column(Integer, nullable=True, default=0)
    api_calls_count = Column(Integer, nullable=True, default=0)
    estimated_cost = Column(String(20), nullable=True)

API Key Management (ApiKey)

The ApiKey model stores encrypted credentials for external services like Anthropic and OpenAI.

class ApiKey(Base):
    __tablename__ = "api_keys"

    id = Column(Integer, primary_key=True, index=True)
    provider = Column(String(100), nullable=False, unique=True, index=True)
    key_value = Column(Text, nullable=False)   # Encrypted in production

    is_active = Column(Boolean, default=True)
    description = Column(Text, nullable=True)
    last_used = Column(DateTime(timezone=True), nullable=True)

Alembic Migration Strategy for Schema Evolution

The project uses Alembic to version database schema changes located in app/backend/alembic/versions/. Each migration script corresponds to a specific model addition or schema modification.

Key Migration Files

Migration File	Target Table	Purpose
5274886e5bee_add_hedgefundflow_table.py	hedge_fund_flows	Creates the flow configuration table with JSON columns for React Flow state.
2f8c5d9e4b1a_add_hedgefundflowrun_table.py	hedge_fund_flow_runs	Establishes the execution tracking table with foreign keys to flows.
3f9a6b7c8d2e_add_hedgefundflowruncycle_table.py	hedge_fund_flow_run_cycles	Adds the cycle detail table and defensive column additions to the runs table.
add_api_keys_table.py	api_keys	Creates the API credential storage with unique provider constraints.

Defensive Migration Patterns

The migrations implement defensive checks to prevent errors on partially migrated databases. For example, in 3f9a6b7c8d2e_add_hedgefundflowruncycle_table.py, the script inspects existing columns before adding new ones:

# Defensive check before adding columns

if 'trading_mode' not in existing_columns:
    op.add_column('hedge_fund_flow_runs', 
                  Column('trading_mode', String(50), nullable=False, default="one-time"))

This pattern ensures idempotent schema migrations that can safely run on databases at different migration states.

Database Connection and FastAPI Integration

The database layer is initialized in app/backend/database/connection.py, which exports the Base declarative class, engine, SessionLocal factory, and the get_db dependency function.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./hedge_fund.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

FastAPI endpoints consume this dependency using Depends(get_db) to ensure proper session lifecycle management.

Practical Implementation Examples

Creating a New Flow Configuration

from app.backend.database.connection import SessionLocal
from app.backend.database import models

def create_flow(name: str, nodes: list, edges: list, viewport: dict = None):
    db = SessionLocal()
    try:
        flow = models.HedgeFundFlow(
            name=name,
            nodes=nodes,
            edges=edges,
            viewport=viewport,
            is_template=False,
            tags=["research"]
        )
        db.add(flow)
        db.commit()
        db.refresh(flow)
        return flow.id
    finally:
        db.close()

Starting a Flow Execution Run

from datetime import datetime
from app.backend.database import models
from app.backend.database.connection import SessionLocal

def start_flow_run(flow_id: int, request_data: dict):
    db = SessionLocal()
    try:
        run = models.HedgeFundFlowRun(
            flow_id=flow_id,
            status="IN_PROGRESS",
            started_at=datetime.utcnow(),
            trading_mode="one-time",
            request_data=request_data
        )
        db.add(run)
        db.commit()
        db.refresh(run)
        return run.id
    finally:
        db.close()

Recording Cycle Analytics and Costs

def record_cycle(flow_run_id: int, cycle_number: int, signals: dict, 
                 decisions: dict, portfolio: dict, llm_calls: int = 0):
    db = SessionLocal()
    try:
        cycle = models.HedgeFundFlowRunCycle(
            flow_run_id=flow_run_id,
            cycle_number=cycle_number,
            started_at=datetime.utcnow(),
            analyst_signals=signals,
            trading_decisions=decisions,
            portfolio_snapshot=portfolio,
            llm_calls_count=llm_calls,
            status="COMPLETED"
        )
        db.add(cycle)
        db.commit()
        return cycle.id
    finally:
        db.close()

Managing External API Keys

def upsert_api_key(provider: str, key_value: str, description: str = ""):
    db = SessionLocal()
    try:
        existing = db.query(models.ApiKey).filter_by(provider=provider).first()
        if existing:
            existing.key_value = key_value
            existing.description = description
        else:
            new_key = models.ApiKey(
                provider=provider,
                key_value=key_value,
                description=description
            )
            db.add(new_key)
        db.commit()
    finally:
        db.close()

Integration with Pydantic Schemas

The backend validates API requests using Pydantic models defined in app/backend/models/schemas.py. These schemas mirror the SQLAlchemy models and include structures like FlowCreateRequest, FlowRunCreateRequest, and ApiKeyCreateRequest, ensuring type safety between HTTP payloads and database operations.

Summary

• Database models and Alembic migrations in the AI Hedge Fund backend provide version-controlled persistence for trading workflows using SQLite and SQLAlchemy.
• Four core models—HedgeFundFlow, HedgeFundFlowRun, HedgeFundFlowRunCycle, and ApiKey—store React Flow configurations, execution metadata, cycle analytics, and API credentials respectively.
• Migration scripts in app/backend/alembic/versions/ use defensive programming patterns to safely evolve the schema without data loss.
• The SessionLocal dependency in app/backend/database/connection.py enables FastAPI endpoints to perform atomic database transactions with automatic session cleanup.
• JSON columns throughout the models allow flexible storage of React Flow state, portfolio snapshots, and analyst signals without requiring rigid schema changes.

Frequently Asked Questions

What database does the AI Hedge Fund backend use?

The backend uses SQLite as its primary database engine, configured in app/backend/database/connection.py with the URL sqlite:///./hedge_fund.db. This choice provides zero-configuration persistence suitable for the project's deployment model while supporting full SQLAlchemy ORM capabilities.

How are database schema changes managed in the project?

Schema changes are managed through Alembic migrations stored in app/backend/alembic/versions/. Each migration script is named with a revision hash and descriptive suffix (e.g., 5274886e5bee_add_hedgefundflow_table.py). The migrations use defensive checks to verify column existence before applying changes, ensuring safe upgrades across different database states.

What is the purpose of the HedgeFundFlowRunCycle model?

The HedgeFundFlowRunCycle model captures granular data from individual analysis-trading cycles within a flow execution. It stores analyst signals, trading decisions, portfolio snapshots, and cost metrics like llm_calls_count and estimated_cost, enabling detailed post-execution analysis of algorithmic trading performance.

Where are the SQLAlchemy models defined?

All SQLAlchemy ORM classes are defined in app/backend/database/models.py. This file contains the HedgeFundFlow, HedgeFundFlowRun, HedgeFundFlowRunCycle, and ApiKey classes that map to the hedge_fund_flows, hedge_fund_flow_runs, hedge_fund_flow_run_cycles, and api_keys tables respectively.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}