finance_data_platform/
├── docker-compose.yml
├── requirements.txt
├── .env.example
│
├── sql/
│   └── sma.sql                    # Calculate SMA5 and SMA20
│   └── daily_returns.sql          # Calculate daily returns
│   └── volatility.sql             # Calculate volatility (standard deviation) of the returns from last 21 trading days
│   └── cumulative_return.sql      # Calculate cumulative returns of the stocks from the earliest OHLCV data of stocks
│
├── data/
│   ├── raw/{symbol}/              # Raw JSON responses from AlphaVantage
│   └── analytics/                 # Extracted files for SQL query
│
├── src/
│   ├── db_connect.py              # psycopg2 connection factory (reads from .env)
│   ├── pipeline.py                # Main ETL orchestration (fetch → transform → load)
│   ├── reprocess_pipeline.py      # Load from local files without API calls
│   ├── analytics.py               # Save the SQL query into .csv file
│   │
│   ├── extract/
│   │   └── alphavantage_ingest.py # API fetch with rate-limit retry and quota detection
│   │
│   ├── transform/
│   │   └── transform_stock.py     # OHLCV + calendar attribute derivation; metadata parsing
│   │
│   ├── load/
│   │   ├── dimension_loader.py    # Bulk-insert dim_date and dim_metadata (ON CONFLICT DO NOTHING)
│   │   └── fact_loader.py         # Bulk-insert stock_prices; get_max_loaded_date for incremental load
│   │
│   └── modeling/
│       ├── create_dimension_tables.py
│       ├── create_fact_tables.py
│       └── create_indexes.py
│
└── tests/
    ├── conftest.py                # db_cursor fixture — isolated TEMP tables per test
    ├── test_transform.py          # Unit tests for transformation logic
    └── test_loaders.py            # Integration tests for all loaders (dim + fact)

# define docker version
version: '3.8'
# define services used in container
services:
  postgres:
    image: postgres:15
    container_name: finance_postgres
    restart: always
    environment:
      POSTGRES_USER: finance_user
      POSTGRES_PASSWORD: finance_password
      POSTGRES_DB: finance_db
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
volumes:
  postgres_data:

docker compose up -d

# verify the container
docker ps
# run the container
docker exec -it finance_postgres psql -U finance_user -d finance_db

# create venv folder
python -m venv venv
# for PowerShell user, we need to set the execution policy before activating the venv
Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy Unrestricted -Force
# activate the venv, for PowerShell users, we use the followings
.\venv\Scripts\Activate.ps1

# install the required packages
pip install requests pandas psycopg2-binary python-dotenv pytest
# save the required packages as requirements
pip freeze > requirements.txt

ALPHAVANTAGE_API_KEY=<alphavantage_api_key_here>
DB_HOST=localhost
DB_PORT=5432
DB_NAME=finance_db
DB_USER=finance_user
DB_PASSWORD=finance_password

.env

import requests
import os
import json
from datetime import datetime
import time
import logging
import pathlib
from dotenv import load_dotenv
# load environment variables from .env file to the codespace
load_dotenv()
# refer API key from .env file
API_KEY = os.getenv('ALPHAVANTAGE_API_KEY')
# define the logger object to inform the progress along the way
logger = logging.getLogger(__name__)
# create the target filepath (where the fetched files are saved into)
RAW_DATA_DIR = pathlib.Path(__file__).parent.parent.parent / "data" / "raw"
# fetch(extract) the OHLCV data
def fetch_stock_prices(symbol):
    # to fetch the data, we need the querying URL
    url = "https://www.alphavantage.co/query"
    # and also the query itself. We need to use the query as dictionary as required from requests.get() method. noted that the dictionary keys must comply with API function parameters.
    params = {
        "function": "TIME_SERIES_DAILY",  # function to call for daily OHLCV data
        "symbol": symbol,  # pass the symbol from method parameters to query parameters
        "apikey": API_KEY,  # refer the loaded API key
        "outputsize": "compact"  # define the output size
    }
    # fetch the data, if it uses more than 60 secs for a request, kill it.
    response = requests.get(url, params=params, timeout=60)
    
    # in case of other responses than 200 (HTTPError), raise it
    response.raise_for_status()
    # save the response as JSON
    data = response.json()
    # AlphaVantage returns the "Information" in case that the request wasn't fulfilled e.g. attempting intraday data with free-tier API keys
    if "Information" in data:
        raise RuntimeError(f"AlphaVantage daily quota exceeded for {symbol}: {data['Information']}")
    # In case exceeding usage quota, AlphaVantage returns "Note" response. So we could wait and retry for a few times.
    if "Note" in data:
        for attempt in range(1, 4):
            logger.warning(f"API rate limit reached for {symbol}. Attempt {attempt}/3 — waiting 60 seconds...")
            time.sleep(60)
            retry_response = requests.get(url, params=params, timeout=60)
            retry_response.raise_for_status()
            retry_data = retry_response.json()
            if "Note" not in retry_data:
                data = retry_data
                break
        else:
            raise RuntimeError(f"API rate limit for {symbol} persisted after 3 retry attempts")
    # In case that request is invalid or cannot be processed, return "Error Message"
    if "Error Message" in data:
        logger.error(f"API error for {symbol}: {data['Error Message']}")
        raise RuntimeError(f"AlphaVantage API error for {symbol}: {data['Error Message']}")
    # create the timestamp for OHLCV data
    timestamp = datetime.utcnow().strftime("%Y-%m-%d")
    # create new blank .json file into target directory. If it's not existed yet, create the new one
    filepath = RAW_DATA_DIR / symbol / f"{symbol}_{timestamp}.json"
    filepath.parent.mkdir(parents=True, exist_ok=True)
    # write the extracted .json response into file, then save close it.
    with open(filepath, "w") as f:
        json.dump(data, f)
    return str(filepath)
# fetch the company metadata
def fetch_company_metadata(symbol):
    # define URL and query parameters as in OHLCV data
    url = "https://www.alphavantage.co/query"
    params = {
        "function": "OVERVIEW",
        "symbol": symbol,
        "apikey": API_KEY
    }
    # fetch the response and store as data
    response = requests.get(url, params=params, timeout=60)
    response.raise_for_status()
    data = response.json()
    # Error Handling
    if "Information" in data:
        raise RuntimeError(f"AlphaVantage daily quota exceeded for {symbol} metadata: {data['Information']}")
    if "Note" in data:
        for attempt in range(1, 4):
            logger.warning(f"API rate limit reached fetching metadata for {symbol}. Attempt {attempt}/3 — waiting 60 seconds...")
            time.sleep(60)
            retry_response = requests.get(url, params=params, timeout=60)
            retry_response.raise_for_status()
            retry_data = retry_response.json()
            if "Note" not in retry_data:
                data = retry_data
                break
        else:
            raise RuntimeError(f"API rate limit for {symbol} metadata persisted after 3 retry attempts")
    if "Error Message" in data:
        raise RuntimeError(f"AlphaVantage API error for {symbol} metadata: {data['Error Message']}")
    # create raw .json file
    filepath = RAW_DATA_DIR / symbol / f"{symbol}_metadata.json"
    filepath.parent.mkdir(parents=True, exist_ok=True)
    # write the response into .json file, then close
    with open(filepath, "w") as f:
        json.dump(data, f)
    return str(filepath)

import json
import pandas as pd
# Transform OHLCV data
def transform_stock_prices(filepath, symbol):
    # read the extracted JSON file
    with open (filepath, "r") as f:
        data = json.load(f)
    # get the dictionary values (the daily OHLCV data) from the "Time Series (Daily)" key in the JSON data >> refer to API documentation for other APIs
    time_series = data.get("Time Series (Daily)", {}) # {} means if the key is not found, it will return an empty dictionary instead of throwing an error
    # create the blank lists, we will append it in the following for loop
    records = []
    for date, values in time_series.items():
        # convert the date string to datetime object, we made it here to reduce the computations overhead than converting it in the loop
        dt = pd.to_datetime(date)
        # select the symbol, date, OHLCV data, then create the new date columns built from date
        records.append({
            "symbol": symbol,   # pass the argument symbol to the function and include it in the records
            "date": dt,       # convert the date string to a datetime object
            "open": float(values["1. open"]),
            "high": float(values["2. high"]),
            "low": float(values["3. low"]),
            "close": float(values["4. close"]),
            "volume": int(values["5. volume"]),
            "day" : dt.day,
            "month" : dt.month,
            "year" : dt.year,
            "quarter" : ((dt.month - 1) // 3) + 1, # ensure the quarter is classified correctly
            "day_of_week" : dt.dayofweek,   # Monday=0, Sunday=6
            "week_of_year" : int(dt.isocalendar().week)   # typecasting to int to avoid pandas Int64 type which is not compatible with psycopg2 when loading into the database
        })
    # convert the list into the dataframe
    df = pd.DataFrame(records)
    return df
# Transform company's metadata
def transform_company_metadata(filepath):
    with open(filepath, "r") as f:
        data = json.load(f)
    # get the dicionary values
    symbol = data.get("Symbol", "")
    if not symbol:
        raise ValueError(f"Metadata file is missing 'Symbol' field. The file may contain an API error response.")
    # this time, we store it as dictionary
    metadata = {
        "symbol": symbol,
        "company_name": data.get("Name", ""),
        "sector": data.get("Sector", "")
    }
    return metadata

# borrow the connection config from db_connect.py
from db_connect import db_connect
def create_fact_table():
    # open the SQL connection
    conn = db_connect()
    # create a cursor object to interact with the database
    cursor = conn.cursor()
    # check the SQL database if they have SQL tables. If not, create the new one.
    cursor.execute(
        """
        CREATE TABLE IF NOT EXISTS stock_prices (
            symbol TEXT,
            date DATE,
            open NUMERIC,
            high NUMERIC,
            low NUMERIC,
            close NUMERIC,
            volume BIGINT,
            PRIMARY KEY (symbol, date),
            FOREIGN KEY (symbol) REFERENCES dim_metadata(symbol),
            FOREIGN KEY (date) REFERENCES dim_date(date)
        );
        """
    )
    # send the query string to SQL
    conn.commit()
    # close both cursor object and connection
    cursor.close()
    conn.close()
# entry point guard to regulate execution sequence in the pipeline
if __name__ == "__main__":
    create_fact_table()

from db_connect import db_connect
def create_dim_dates():
    # open the SQL connection, it requires host, port, database_name, username, and password
    conn = db_connect()
    # create a cursor object to interact with the database
    cursor = conn.cursor()
    # create the dim_dates table if not exists, otherwise do nothing
    cursor.execute(
        """
        CREATE TABLE IF NOT EXISTS dim_date (
            date DATE PRIMARY KEY,
            day INT,
            month INT,
            year INT,
            quarter INT,
            day_of_week INT,
            week_of_year INT
        );
        """
    )
    conn.commit()  # commit the transaction to save changes
    cursor.close()  # close the cursor
    conn.close()  # close the connection
def create_dim_metadata():
    conn = db_connect()
    cursor = conn.cursor()
    cursor.execute(
        """
        CREATE TABLE IF NOT EXISTS dim_metadata (
            symbol TEXT PRIMARY KEY,
            company_name TEXT,
            sector TEXT
            );
        """       
    )
    conn.commit()   # 
    cursor.close()
    conn.close()
# entry point guard
if __name__ == "__main__":
    create_dim_dates()
    create_dim_metadata()

from psycopg2.extras import execute_values
import logging
logger = logging.getLogger(__name__)
# open the connection to the PostgreSQL database using credentials from environment variables
def load_stock_prices(cursor, df):
    
    # insert data into stock_prices table as a bulk
    # Convert all numpy types to Python native types to avoid psycopg2 adapter errors
    fact_columns = ["symbol", "date", "open", "high", "low", "close", "volume"] # select the columns
    df_copy = df[fact_columns].copy()  # create a copy with selected columns to avoid modifying the original DataFrame
    df_copy['date'] = df_copy['date'].dt.to_pydatetime()  # numpy.datetime64 -> datetime
    
    # Use .values.tolist() to convert numpy types to native Python types
    values = [tuple(row) for row in df_copy[fact_columns].values.tolist()]
    # write the query to load the transformed values to SQL table
    insert_query = """
        INSERT INTO stock_prices (symbol, date, open, high, low, close, volume)
        VALUES %s
        ON CONFLICT (symbol, date) DO NOTHING
    """
    # load the value into table as a bulk
    execute_values(cursor, insert_query, values)
    logger.info(f"Loaded {len(values)} new rows into stock_prices")
def get_max_loaded_date(cursor, symbol):
    """
    Check for the latest date loaded for each symbol in fact table.
    """
    cursor.execute(
        "SELECT MAX(date) FROM stock_prices WHERE symbol = %s;",
        (symbol,)
    )
    # fetchone() returns a tuple, we want the first element which is the max date
    result = cursor.fetchone()
    max_date = result[0]
    logger.info(f"Latest loaded date for {symbol}: {max_date}")
    return max_date

from psycopg2.extras import execute_values
import logging
# create logging object to return the execution log to user
logger = logging.getLogger(__name__)
def load_dim_dates(cursor, df):
    """
    Load the date dimensions in bulk,
    reduce the load time significantly than the iteration row-by-row
    """
    date_cols = ['date', 'day', 'month', 'year', 'quarter', 'day_of_week', 'week_of_year']
    df = df[date_cols].drop_duplicates()
    values = [tuple(row) for row in df.values.tolist()] # convert each low to the list of tuples so it could be loaded as a bulk
    insert_query = """
        INSERT INTO dim_date (date, day, month, year, quarter, day_of_week, week_of_year)
        values %s
        ON CONFLICT (date) DO NOTHING;
    """
    # load values as a bulk into SQL tables
    execute_values(cursor, insert_query,values)
    logger.info(f"Loaded {len(values)} rows into dim_date")
def load_dim_metadata(cursor, metadata):
    # insert one row (metadata) into SQL metadata table
    cursor.execute(
        """
        INSERT INTO dim_metadata (symbol, company_name, sector)
        VALUES (%s, %s, %s)
        ON CONFLICT (symbol) DO NOTHING;
        """,
        (metadata["symbol"], metadata["company_name"], metadata["sector"])
    )
    logger.info(f"Loaded metadata for {metadata['symbol']} into dim_metadata")

import logging
import time
from db_connect import db_connect
from extract.alphavantage_ingest import fetch_stock_prices, fetch_company_metadata
from transform.transform_stock import transform_stock_prices, transform_company_metadata
from load.fact_loader import load_stock_prices, get_max_loaded_date
from load.dimension_loader import load_dim_dates, load_dim_metadata
from modeling.create_dimension_tables import create_dim_dates, create_dim_metadata
from modeling.create_fact_tables import create_fact_table
from modeling.create_indexes import create_indexes
# config the logging to display info level messages with timestamps
logging.basicConfig(
    level=logging.INFO,
    format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt = '%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)
def run(symbols):
    # build up the connection with the borrowed scripts from cb_connect.py
    conn = db_connect()
    cursor = conn.cursor()  # create a cursor object to interact with the database
    for symbol in symbols:
        try:
            # fetch the data from the API, we use 12 secs sleep after each API call to respect free tier limit (5 calls per minute)
            logger.info(f"Fetching {symbol}")
            filepath = fetch_stock_prices(symbol)
            time.sleep(12)
            metadata_filepath = fetch_company_metadata(symbol)
            time.sleep(12)
            # transform the data and metadata
            logger.info("Transforming")
            df = transform_stock_prices(filepath, symbol)
            metadata = transform_company_metadata(metadata_filepath)
            # add the incremental loading logic here by checking the max date already loaded for the symbol in the fact table
            max_date = get_max_loaded_date(cursor, symbol)
            if max_date is not None:
                df = df[df["date"].dt.date > max_date]  # filter the dataframe to only include rows with date greater than max_date
                logger.info(f"{symbol}: {len(df)} new rows after {max_date}")
            if df.empty:
                logger.info(f"{symbol}: no new data to load, skipping")
                continue
            # load the dimension tables first, as the fact table has foreign key associated to them
            logger.info("Populating dimension tables")
            load_dim_dates(cursor, df)            
            load_dim_metadata(cursor, metadata)
            # load the fact table with transformed data
            logger.info("Loading into fact table")
            load_stock_prices(cursor, df)
            conn.commit()
        # in case anything goes wrong for each stocks, we cancelled out the commit for that stock, then raise an error
        except Exception as e:
            conn.rollback()
            logger.error(f"Pipeline failed: {e}")
            raise
# Orchestrate all processes together
if __name__ == "__main__":
    symbols = ["NVDA","AAPL","MSFT","GOOGL","AMZN"]
    create_dim_dates()   # ensure the dim_dates table is created before loading data
    create_dim_metadata()  # ensure the dim_metadata table is created before loading data
    create_fact_table()  # ensure the fact table is created before loading data
    create_indexes()  # create indexes on the fact table for performance
    run(symbols)

import pytest
import psycopg2
from src.db_connect import db_connect
# scope the integration test. We will mention it in the integration test, and we could also scope only the integration test later in the terminal.
def pytest_configure(config):
    config.addinivalue_line(
        "markers", "integration: mark a test as an integration test that requiring a live PostgreSQL connection (docker compose up -d)"
    )
# set the figure, we would use it now and later when the test loaders need the table
@pytest.fixture(scope="function")
def db_cursor():
    """
    Connect to the real Postgres instance and create TEMP tables
    , which mimics the production schema. Each test gets a clean slate;
    connection is rolled back and closed after the test regardless of pass/fail.
    Skip automatically if Postgres is not reachable.
    """
    try:
        conn = db_connect()
    except psycopg2.OperationalError:
        pytest.skip("PostgreSQL is not reachable, compose the docker container first via: docker compose up -d")
    conn.autocommit = False
    cursor = conn.cursor()
    """
    TEMP tables are session-scoped and don't support cross-table foreign keys (FK) constraints,
    so they are defined without them here. They are auto-dropped when the connection is closed .
    """
    # create temp date dimension table
    cursor.execute(
        """
        CREATE TEMP TABLE dim_date (
            date            DATE PRIMARY KEY,
            day             INT,
            month           INT,
            year            INT,
            quarter         INT,
            day_of_week     INT,
            week_of_year    INT
        );
        """
    )
    # create temp metadata dimension table
    cursor.execute(
        """
        CREATE TEMP TABLE dim_metadata (
            symbol         TEXT PRIMARY KEY,
            company_name   TEXT,
            sector         TEXT
        );
        """
    )
    # create temp fact table
    cursor.execute(
        """
        CREATE TEMP TABLE stock_prices (
            symbol         TEXT,
            date           DATE,
            open           NUMERIC,
            high           NUMERIC,
            low            NUMERIC,
            close          NUMERIC,
            volume         BIGINT,
            PRIMARY KEY (symbol, date)
        );
        """
    )
    yield cursor    # yield the cursor to the test function. This was used to separate the setup and teardown logic, which is a best practice for test fixtures.
    conn.rollback() # rollback any changes to ensure test isolation
    cursor.close()
    conn.close()

import pytest
import pandas as pd
from datetime import datetime
import json
import tempfile
import os
from src.transform.transform_stock import transform_stock_prices, transform_company_metadata
class TestTransformStockPrices:
  ...
class TestTransformCompanyMetadata:
  @pytest.fixture
    def sample_metadata_response(self):
        """
        Create a mock metadata API response to interact with the following testing methods.
        """
        return {
            "Symbol": "AAPL",
            "Name": "Apple Inc.",
            "Sector": "Technology",
            "Industry": "Consumer Electronics"
        }
    
    def test_transform_metadata_returns_dict(self, sample_metadata_response):
        """
        Ensure that transformed data is a dictionary.
        """
        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
            json.dump(sample_metadata_response, f)
            temp_filepath = f.name
        try:
            result = transform_company_metadata(temp_filepath)
            assert isinstance(result, dict)
        finally:
            os.unlink(temp_filepath)

import pytest
import pandas as pd
from datetime import date
from src.load.fact_loader import load_stock_prices, get_max_loaded_date
from src.looad.dimension_loader import load_dim_dates, load_dim_metadata
# define the helper functions for further testing as show below
def _dim_date_row(date_str):
    dt = pd.Timestamp(date_str)
    return {
        "date": dt,
        "day": dt.day,
        "month": dt.month,
        "year": dt.year,
        "quarter": ((dt.month - 1) // 3) + 1,
        "day_of_week": dt.dayofweek,
        "week_of_year": int(dt.isocalendar().week)
    }
  
def _price_row(symbol, date_str, opening=148.0, closing=150.0):
    dt = pd.Timestamp(date_str)
    return {
        "symbol": symbol,
        "date": dt,
        "open": opening,
        "high": closing + 1,
        "low":  opening - 1,
        "close": closing,
        "volume": 25000000,
        "day": dt.day,
        "month": dt.month,
        "year": dt.year,
        "quarter": ((dt.month - 1) // 3) + 1,
        "day_of_week": dt.dayofweek,
        "week_of_year": int(dt.isocalendar().week)
    }
# create the class of testing methods
@pytest.mark.integration
class TestLoadDimDates:
  ...
@pytest.mark.integration
class TestLoadStockPrices:
    def _seed(self, db_cursor, dates=("2026-03-09", "2026-03-08"), symbol="AAPL"):
        """
        Helper method to seed the dim_date and dim_metadata tables with the necessary data for testing the fact loader.
        """
        dim_df = pd.DataFrame([_dim_date_row(d) for d in dates])
        load_dim_dates(db_cursor, dim_df)
        load_dim_metadata(db_cursor, {"symbol": symbol, "company_name": "Apple Inc.", "sector": "TECHNOLOGY"})
    def test_inserts_rows(self, db_cursor):
        """
        Given the blank stock_prices fact table,
        when we load a stock price record,
        then it should be inserted into the table in just one row.
        """
        self._seed(db_cursor)
        df = pd.DataFrame([_price_row("AAPL", "2026-03-09"), _price_row("AAPL", "2026-03-08")])
        load_stock_prices(db_cursor, df)
        db_cursor.execute("SELECT COUNT(*) FROM stock_prices;")
        assert db_cursor.fetchone()[0] == 2

pytest
  ├── discovers tests/conftest.py → registers "integration" marker
  ├── test_transform.py (unit tests)
  │     └── no db_cursor → runs immediately, no Docker needed
  └── test_loaders.py (integration tests)
        └── requests db_cursor fixture
              ├── tries db_connect()
              ├── if Docker is down → pytest.skip()
              └── if Docker is up → creates TEMP tables → yields cursor → test runs → rollback → cleanup

# use this one to run everything
pytest tests/ -v
# use this one to only run on the integration tests
# make sure the docker PostgreSQL container is running. Otherwise run docker compose up -d
pytest tests/ -v -m integration

SELECT
    s.symbol,
    s.date,
    s.close,
    LAG(s.close) OVER (
        PARTITION BY s.symbol
        ORDER BY s.date ASC
        ) AS prev_close,
    ROUND (
        (s.close / LAG(s.close) OVER ( PARTITION BY s.symbol ORDER BY s.date ASC)) - 1, 4
    ) AS daily_return
    FROM stock_prices s
ORDER BY s.symbol, s.date ASC;

import pandas as pd
from pathlib import Path
from db_connect import db_connect
import logging
from sqlalchemy import create_engine
from dotenv import load_dotenv
import os
load_dotenv()
logger = logging.getLogger(__name__)
# create SQLAlchemy engine for database connection using connection string
engine = create_engine(
    f"postgresql+psycopg2://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"
)
# lookup the queries from /sql
QUERIES = {
    "sma": Path("sql/sma.sql"),
    "daily_returns": Path("sql/daily_returns.sql"),
    "volatility": Path("sql/volatility.sql"),
    "cumulative_return": Path("sql/cumulative_return.sql")
}
# create output directory to save the analytics results
OUTPUT_DIR = Path("data/analytics")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# connect to the database and execute each query, saving the results as CSV files
conn = db_connect()
# execute each query and save the results to CSV files
for name, sql_path in QUERIES.items():
    query = sql_path.read_text()
    df = pd.read_sql_query(query, conn)
    output = OUTPUT_DIR / f"{name}.csv"
    df.to_csv(output, index=False)
    logger.info(f"Exported {name} to {output}")
# close the database connection
engine.dispose()