Quantitative Investment Analytics ETL Pipeline

CTEs and ROW_NUMBER select month-end NAV; LAG computes compounded monthly and 20-session returns; DENSE_RANK ranks assets by trailing volatility. Sector reporting joins positions to the security dimension. The first monthly return is undefined without a preceding month-end.

-- Latest allocations, with sector names obtained from the security dimension.
CREATE VIEW latest_sector_exposure AS
WITH latest AS (SELECT MAX(date) AS date FROM portfolio_positions)
SELECT p.portfolio_id, s.sector, SUM(p.market_value) AS market_value,
       SUM(p.weight) AS weight
FROM portfolio_positions p
JOIN securities s ON s.ticker = p.ticker
JOIN latest l ON l.date = p.date
GROUP BY p.portfolio_id, s.sector;

-- Compound monthly returns from month-end NAV using a window function.
CREATE VIEW monthly_portfolio_returns AS
WITH ranked AS (
 SELECT *, SUBSTR(date, 1, 7) AS month,
 ROW_NUMBER() OVER (PARTITION BY portfolio_id, SUBSTR(date, 1, 7) ORDER BY date DESC) AS rn
 FROM portfolio_daily_summary
), endpoints AS (
 SELECT portfolio_id, month, nav FROM ranked WHERE rn = 1
)
SELECT portfolio_id, month, nav,
 nav / LAG(nav) OVER (PARTITION BY portfolio_id ORDER BY month) - 1 AS monthly_return
FROM endpoints;

-- Daily cross-sectional ranking by trailing annualized volatility.
CREATE VIEW volatility_ranking AS
SELECT date, ticker, rolling_volatility_20,
 DENSE_RANK() OVER (PARTITION BY date ORDER BY rolling_volatility_20) AS volatility_rank
FROM security_daily_analytics WHERE rolling_volatility_20 IS NOT NULL;

CREATE VIEW rolling_portfolio_returns AS
SELECT date, portfolio_id,
 nav / LAG(nav, 20) OVER (PARTITION BY portfolio_id ORDER BY date) - 1 AS return_20_sessions
FROM portfolio_daily_summary;

The selected Ridge penalty is the largest value in the tested grid. A broader penalty range is a useful follow-up, using a new validation plan before inspecting additional test results. Lasso has slightly lower held-out error than Ridge for Low volatility, but the report retains Ridge because selection was based on development CV, not test performance.

Grid search scores raw predictions; final test predictions are clipped to nonnegative volatility. This is a documented evaluation mismatch to align in a future revision. There is no uncertainty interval or significance test in the current experiment.

import numpy as np
import pandas as pd
from sklearn.linear_model import ElasticNet, Lasso, LinearRegression, Ridge
from sklearn.metrics import mean_squared_error, r2_score
from sklearn.model_selection import GridSearchCV, TimeSeriesSplit
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler


FEATURES = ["return_1", "return_5", "return_20", "volatility_20", "market_return_20", "volume_ratio"]


def feature_frame(group, benchmark, volume, horizon):
    r = group.set_index("date").net_return.sort_index()
    frame = pd.DataFrame({"return_1": r, "return_5": (1 + r).rolling(5).apply(np.prod, raw=True) - 1,
                          "return_20": (1 + r).rolling(20).apply(np.prod, raw=True) - 1,
                          "volatility_20": r.rolling(20).std() * np.sqrt(252),
                          "market_return_20": (1 + benchmark).rolling(20).apply(np.prod, raw=True) - 1,
                          "volume_ratio": volume / volume.rolling(20).mean()})
    # The label at t contains returns t+1 through t+h; purge h rows at every split.
    frame["target"] = r.rolling(horizon).std().shift(-horizon) * np.sqrt(252)
    frame["target_end"] = pd.Series(r.index, index=r.index).shift(-horizon)
    return frame.replace([np.inf, -np.inf], np.nan).dropna()


def fit_models(daily, prices, config):
    benchmark = daily[daily.portfolio_id == "Benchmark"].set_index("date").net_return
    volume = prices[prices.ticker == config["benchmark"]].set_index("date").volume
    horizon = config["forecast_days"]
    scores, predictions, coefficients, audits = [], [], [], []
    candidates = {"Linear regression": (LinearRegression(), {}),
                  "Ridge": (Ridge(), {"model__alpha": [0.1, 1., 10., 100.]}),
                  "Lasso": (Lasso(max_iter=20000), {"model__alpha": [0.00001, 0.0001, 0.001]}),
                  "ElasticNet": (ElasticNet(max_iter=20000),
                                 {"model__alpha": [0.0001, 0.001], "model__l1_ratio": [0.25, 0.75]})}
    for name, group in daily.groupby("portfolio_id"):
        if name == "Benchmark":
            continue
        frame = feature_frame(group, benchmark, volume, horizon)
        split = int(len(frame) * (1 - config["test_fraction"]))
        train, test = frame.iloc[:split - horizon], frame.iloc[split:]
        if len(train) < 150 or len(test) < 30:
            raise ValueError("Insufficient history for purged chronological validation")
        if train.target_end.max() >= test.index.min():
            raise ValueError("Training labels overlap the test period")
        cv = TimeSeriesSplit(n_splits=4, gap=horizon)
        for fold, (tr, va) in enumerate(cv.split(train)):
            if train.iloc[tr].target_end.max() >= train.iloc[va].index.min():
                raise ValueError("Cross-validation label leakage")
            audits.append({"portfolio_id": name, "fold": fold, "train_start": train.iloc[tr].index.min(),
                           "last_train_label": train.iloc[tr].target_end.max(),
                           "validation_start": train.iloc[va].index.min(), "validation_end": train.iloc[va].index.max()})
        models = {}
        for model_name, (model, grid) in candidates.items():
            search = GridSearchCV(Pipeline([("scale", StandardScaler()), ("model", model)]), grid,
                                  cv=cv, scoring="neg_mean_squared_error", n_jobs=1)
            search.fit(train[FEATURES], train.target)
            models[model_name] = (np.maximum(search.predict(test[FEATURES]), 0), -search.best_score_, str(search.best_params_))
            for feature, value in zip(FEATURES, search.best_estimator_.named_steps["model"].coef_):
                coefficients.append({"portfolio_id": name, "model": model_name,
                                     "feature": feature, "coefficient": value})
        selected = min(models, key=lambda key: models[key][1])
        models["Persistence baseline"] = (test.volatility_20.to_numpy(), np.nan, "Trailing 20-day volatility")
        for model_name, (prediction, cv_mse, params) in models.items():
            scores.append({"portfolio_id": name, "model": model_name,
                           "rmse": np.sqrt(mean_squared_error(test.target, prediction)),
                           "r2": r2_score(test.target, prediction), "cv_mse": cv_mse,
                           "selected_by_cv": model_name == selected, "parameters": params,
                           "train_end": train.index[-1], "last_train_label": train.target_end.max(),
                           "test_start": test.index[0], "test_end": test.index[-1], "test_rows": len(test)})
            predictions.extend({"date": date, "portfolio_id": name, "model": model_name,
                                "actual": actual, "prediction": value}
                               for date, actual, value in zip(test.index, test.target, prediction))
    return pd.DataFrame(scores), pd.DataFrame(predictions), pd.DataFrame(coefficients), pd.DataFrame(audits)

import sqlite3

import numpy as np
import pandas as pd
import pytest

from src.analytics import metrics, portfolio_analytics
from src.load import load_mart
from src.models import feature_frame, FEATURES
from src.validation import clean_inputs, validate_outputs


@pytest.fixture
def inputs():
    dates = pd.bdate_range("2020-01-01", periods=360)
    frames = []
    for i, ticker in enumerate(["A", "B", "M"]):
        r = 0.0002 + (i + 1) * 0.002 * np.sin(np.arange(len(dates)) / 7)
        frames.append(pd.DataFrame({"date": dates, "ticker": ticker,
                                    "adjusted_price": 100 * np.cumprod(1 + r), "volume": 1000}))
    prices = pd.concat(frames, ignore_index=True)
    holdings = pd.DataFrame({"portfolio_id": ["Balanced"] * 2, "ticker": ["A", "B"], "target_weight": [0.6, 0.4]})
    securities = pd.DataFrame({"ticker": ["A", "B", "M"], "sector": ["Tech", "Banks", "Benchmark"], "currency": ["CAD"] * 3})
    config = {"warmup_days": 30, "benchmark": "M", "initial_capital": 100000,
              "transaction_cost_bps": 10, "risk_free_rate": 0.03}
    return prices, holdings, securities, config


def test_compounding_and_initial_loss_drawdown():
    result = metrics([-0.1, 0.1], 0)
    assert result["cumulative_return"] == pytest.approx(-0.01)
    assert result["max_drawdown"] == pytest.approx(-0.1)
    assert result["annualized_return"] == pytest.approx(0.99 ** 126 - 1)


def test_zero_volatility_sharpe_is_undefined():
    assert np.isnan(metrics([0., 0., 0.])["sharpe"])


@pytest.mark.parametrize("fault", ["negative", "missing", "infinite", "conflict", "gap", "weights", "unknown"])
def test_reject_bad_inputs(inputs, fault):
    prices, holdings, securities, _ = inputs
    if fault == "negative":
        prices.loc[0, "adjusted_price"] = -1
    elif fault == "missing":
        prices.loc[0, "volume"] = np.nan
    elif fault == "infinite":
        prices.loc[0, "adjusted_price"] = np.inf
    elif fault == "conflict":
        prices = pd.concat([prices, prices.iloc[:1].assign(adjusted_price=999)])
    elif fault == "gap":
        prices = prices.iloc[1:]
    elif fault == "weights":
        holdings.loc[0, "target_weight"] = 0.8
    elif fault == "unknown":
        holdings.loc[0, "ticker"] = "UNKNOWN"
    with pytest.raises(ValueError):
        clean_inputs(prices, holdings, securities)


def test_exact_duplicates_are_audited(inputs):
    p, h, s, _ = inputs
    cleaned, audit = clean_inputs(pd.concat([p, p.iloc[:1]]), h, s)
    assert len(cleaned) == len(p)
    assert audit["exact_duplicates_removed"] == 1


def test_previous_weights_and_reconciliation(inputs):
    p, h, s, c = inputs
    daily, positions, *_ = portfolio_analytics(p, h, s, c)
    validate_outputs(daily, positions)
    wide = p.pivot(index="date", columns="ticker", values="adjusted_price")
    r = wide.pct_change().iloc[31]
    observed = daily[daily.portfolio_id == "Balanced"].iloc[1]
    assert observed.gross_return == pytest.approx(0.6 * r.A + 0.4 * r.B)
    assert daily[daily.portfolio_id == "Benchmark"].cost.sum() == 0


def test_costs_reduce_nav(inputs):
    p, h, s, c = inputs
    cost = portfolio_analytics(p, h, s, c)[0]
    free = portfolio_analytics(p, h, s, dict(c, transaction_cost_bps=0))[0]
    a = cost[cost.portfolio_id == "Balanced"]
    b = free[free.portfolio_id == "Balanced"]
    assert a.cost.sum() > 0
    assert a.nav.iloc[-1] < b.nav.iloc[-1]


def test_low_vol_weights_ignore_future_prices(inputs):
    p, h, s, c = inputs
    first = portfolio_analytics(p, h, s, c)[-1]
    altered = p.copy()
    mask = altered.date > sorted(p.date.unique())[c["warmup_days"]]
    altered.loc[mask & (altered.ticker == "A"), "adjusted_price"] *= 2
    second = portfolio_analytics(altered, h, s, c)[-1]
    pd.testing.assert_frame_equal(first, second)


def test_forecast_label_and_features_use_correct_dates():
    dates = pd.bdate_range("2020-01-01", periods=120)
    r = pd.Series(np.sin(np.arange(120)) * 0.01, index=dates)
    group = pd.DataFrame({"date": dates, "net_return": r.values})
    frame = feature_frame(group, r, pd.Series(1000, index=dates), 20)
    t = frame.index[10]
    i = dates.get_loc(t)
    assert frame.loc[t, "target"] == pytest.approx(r.iloc[i + 1:i + 21].std() * np.sqrt(252))
    altered = group.copy()
    altered.loc[altered.date > t, "net_return"] = 0.2
    after = feature_frame(altered, r, pd.Series(1000, index=dates), 20)
    np.testing.assert_allclose(frame.loc[t, FEATURES].astype(float), after.loc[t, FEATURES].astype(float))


def test_atomic_load_preserves_previous_database(tmp_path):
    with sqlite3.connect(tmp_path / "investment_analytics.db") as conn:
        conn.execute("CREATE TABLE original (value INTEGER)")
        conn.execute("INSERT INTO original VALUES (7)")
    with pytest.raises(sqlite3.Error):
        load_mart(tmp_path, {"new_table": pd.DataFrame({"x": [1]})}, "INVALID SQL;")
    with sqlite3.connect(tmp_path / "investment_analytics.db") as conn:
        assert conn.execute("SELECT value FROM original").fetchone()[0] == 7

{
  "start": "2018-01-01",
  "end": "2026-09-01",
  "benchmark": "XIC.TO",
  "initial_capital": 100000,
  "risk_free_rate": 0.03,
  "transaction_cost_bps": 10,
  "warmup_days": 252,
  "forecast_days": 20,
  "test_fraction": 0.2,
  "seed": 42
}