Nepali investors and fund managers scaling beyond basic asset allocations frequently hit a wall: manual spreadsheets fail when confronting irregular trading calendars, sudden ticker delistings, and unsynchronized price adjustments. This ad-hoc approach is not only prone to human error but also scales poorly, leaving critical investment decisions vulnerable to stale data. I recently tackled this challenge head-on, building a fault-tolerant, vectorized Python pipeline that ingests raw, messy multi-asset time series, dynamically adjusts for corporate events, and computes institutional-grade performance metrics without memory bloat or lookahead bias. This post will walk you through the engineering decisions and practical code needed to construct such a system, showing you how to normalize disparate time series, compute critical financial indicators, and confidently evaluate investment strategies. The core judgment I came to is that a well-engineered data pipeline is not a luxury, but a necessity for informed financial decision-making in today's dynamic markets.
Key Takeaways
- Why standard outer joins on disparate time-series destroy portfolio variance-covariance matrices and how forward-fill strategies must account for market holidays.
- Implementing vectorized rolling window calculations for Sharpe ratio and maximum drawdown using pure NumPy/Pandas without explicit
forloops. - Designing defensive schema validation guards against missing API payloads, zero-division hazards, and malformed float strings.
- Structuring a modular ingestion pipeline that isolates network IO limits from compute-heavy quantitative transformations.
The Problem
In my previous work building a mutual fund performance engine, we touched upon automating performance tracking. However, when you move beyond a single fund or a handful of synchronized assets to a diverse portfolio spanning multiple markets and asset classes, the complexity explodes. Manual spreadsheets, even with sophisticated macros, quickly become a brittle house of cards. Irregular trading calendars across different exchanges, the sudden delisting of a ticker, or the subtle impact of corporate actions like stock splits or dividends (which often aren't immediately reflected in simple price feeds) can introduce insidious errors. These errors don't just affect historical accuracy; they can lead to misinformed investment decisions, especially in volatile markets where every basis point counts. The challenge is to build a system that can gracefully handle these real-world data imperfections, provide consistent performance metrics, and do it all efficiently at scale.
Data and Sources
For this project, we'll leverage the yfinance Python client to query historical daily open-high-low-close-volume (OHLCV) datasets from Yahoo Finance. This provides a robust source for cross-border equities and benchmark indices. We'll be pulling data for common equities like AAPL, MSFT, GOOG, and a benchmark index like ^GSPC (S&P 500) over a multi-year window to demonstrate the time-series challenges.
yfinanceofficial documentation: https://pypi.org/project/yfinance/- Modern Portfolio Theory (MPT) foundational equations for covariance matrix shrinkage and annualized volatility scaling: https://www.investopedia.com/terms/m/modernportfoliotheory.asp
- Pandas time-series offset aliases and robust resampling patterns: https://pandas.pydata.org/pandas-docs/stable/user_guide/timeseries.html#timeseries-offset-aliases
Data accessed on 2024-05-15.
Step 1 — Resilient Multi-Asset Ingestion and Schema Harmonization
The first hurdle in building any robust financial data pipeline is reliable data ingestion. External APIs, like Yahoo Finance, can be flaky – temporary network issues, rate limits, or even upstream data provider problems can lead to partial network drops or missing data for specific tickers. Our goal here is to pull parallel asset histories without failing catastrophically on these transient issues, and then to harmonize their schemas into a consistent, easily digestible format.
I typically use a retry mechanism to make API calls more resilient. For Python, tenacity is an excellent library for this. We'll wrap our yfinance.download calls to automatically retry on common network errors. Once data is retrieved, yfinance returns a Pandas MultiIndex DataFrame, which can be cumbersome. We need to flatten this into a single DataFrame with a clean date index and clearly identified columns for each asset, specifically focusing on the 'Close' price for our calculations.
import yfinance as yf
import pandas as pd
from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type
import requests
@retry(stop=stop_after_attempt(3), wait=wait_fixed(5),
retry=retry_if_exception_type(requests.exceptions.RequestException))
def fetch_ticker_data(ticker, start_date, end_date):
"""
Fetches historical OHLCV data for a single ticker with retry logic.
"""
print(f"Fetching data for {ticker} from {start_date} to {end_date}...")
try:
data = yf.download(ticker, start=start_date, end=end_date, progress=False)
if data.empty:
print(f"Warning: No data returned for {ticker}.")
return data
except Exception as e:
print(f"Error fetching {ticker}: {e}")
raise # Re-raise to trigger tenacity retry
def resilient_multi_asset_ingestion(tickers, start_date, end_date):
"""
Ingests data for multiple tickers, handles retries, and harmonizes schema.
"""
all_data = {}
for ticker in tickers:
try:
df = fetch_ticker_data(ticker, start_date, end_date)
if not df.empty:
all_data[ticker] = df['Close'].rename(ticker)
except Exception as e:
print(f"Failed to fetch {ticker} after retries: {e}. Skipping.")
# Combine all series into a single DataFrame
if not all_data:
raise ValueError("No data could be fetched for any tickers.")
combined_df = pd.concat(all_data.values(), axis=1)
return combined_df
The fetch_ticker_data function uses tenacity to retry up to 3 times with a 5-second delay if a requests.exceptions.RequestException occurs. This covers most network-related issues. The resilient_multi_asset_ingestion function then iterates through the provided tickers, collecting their 'Close' prices and renaming the series to the ticker symbol for clarity. Finally, it concatenates these series into a single DataFrame, where each column represents a ticker's closing price, indexed by date.
Step 2 — Handling Market Calendars and Missing Data
Once we have the raw price data, the real challenge begins: aligning asynchronous trading days across international assets. Different markets have different holidays, and some assets might have been listed later than others (IPOs). A naive outer join or simple forward-fill can introduce lookahead bias or distort correlations. My approach is to construct a "master" trading calendar and then carefully reindex and fill missing values.
First, we create a union of all dates present in our fetched data. Then, we reindex each asset's price series to this master calendar. For gaps due to market holidays, we apply a strict forward-fill (ffill) with a `limit` parameter. This `limit` is crucial: it prevents prices from being carried forward indefinitely if an asset is delisted or truly stops trading, which would introduce significant lookahead bias. For initial `NaN` values (e.g., an asset's IPO date is after the portfolio start date), a `bfill` (backward-fill) is appropriate only if we're careful about the context, otherwise, it's safer to leave them as `NaN` until actual trading begins.
def align_and_fill_data(combined_df, start_date, end_date):
"""
Aligns time series to a common date range, handles missing data with ffill/bfill.
"""
# Create a complete date range for alignment
full_date_range = pd.date_range(start=start_date, end=end_date, freq='B') # Business days
# Reindex the combined DataFrame to the full business day range
aligned_df = combined_df.reindex(full_date_range)
# Forward-fill missing values, limit to prevent carrying stale data too far
# A limit of 5 business days covers a week of holidays/non-trading days
aligned_df = aligned_df.ffill(limit=5)
# Backward-fill any remaining NaNs at the beginning (e.g., for IPOs)
# This assumes we want to propagate the first available price backwards
# but be cautious as this can introduce bias if not handled carefully.
# For a portfolio, it's often better to start the portfolio value calculation
# only from the first date all assets are available or acquired.
aligned_df = aligned_df.bfill(limit=5) # Propagate first valid observation backwards
# Drop any rows that still have NaNs (e.g., if an asset was never traded in the period)
# Or, more robustly, drop columns (assets) that are entirely NaN after filling
aligned_df = aligned_df.dropna(how='all')
print(f"Aligned data shape: {aligned_df.shape}")
return aligned_df
By reindexing to a full business day range and applying `ffill` with a `limit`, we ensure that our prices reflect the last known value without making unrealistic assumptions about future data. The `bfill` is used sparingly, primarily for the very beginning of a series to ensure an initial price point if available, but it's crucial to understand its implications for lookahead bias when applying it more broadly.
Step 3 — Vectorized Portfolio Return and Risk Engine
With a clean, aligned dataset, we can now build the core of our portfolio engine. The key here is vectorization. Explicit `for` loops in Pandas or NumPy are performance killers for large datasets. We want to leverage their optimized C-backed operations for speed and efficiency.
We'll calculate daily returns for each asset, then use matrix dot products to compute weighted portfolio returns. Portfolio value will be tracked cumulatively. For risk metrics, we'll implement vectorized rolling window calculations for Sharpe ratio and maximum drawdown. The Sharpe ratio requires annualized volatility and returns, while maximum drawdown involves tracking the cumulative maximum value and the subsequent percentage drop.
import numpy as np
def calculate_portfolio_metrics(aligned_prices, portfolio_holdings, benchmark_prices):
"""
Calculates weighted portfolio returns, value, Sharpe ratio, and Max Drawdown.
"""
# Calculate daily returns for all assets
daily_returns = aligned_prices.pct_change().dropna()
# Filter daily_returns to only include dates where all assets in portfolio_holdings exist
# This also naturally handles acquisition dates for portfolio value calculation
portfolio_start_date = min(pd.to_datetime(h['acquisition_date']) for h in portfolio_holdings)
daily_returns = daily_returns[daily_returns.index >= portfolio_start_date]
# Initialize portfolio value and weights
portfolio_value = pd.Series(0.0, index=daily_returns.index)
# Calculate initial portfolio value and weights dynamically
initial_capital = 0
current_weights = {} # Weights will change over time for rebalancing
# Determine the first date where all acquired assets are present
first_valid_portfolio_date = daily_returns.index[0]
for holding in portfolio_holdings:
ticker = holding['ticker']
shares = holding['shares_acquired']
acq_date = pd.to_datetime(holding['acquisition_date'])
# Find the actual price on or after acquisition date
# Use .loc for precise date lookup, .ffill() to get price on first trading day if acq_date is holiday
acq_price_series = aligned_prices.loc[acq_date:, ticker].ffill().dropna()
if acq_price_series.empty:
print(f"Warning: No valid price for {ticker} on or after {acq_date}. Skipping.")
continue
acq_price = acq_price_series.iloc[0]
initial_capital += shares * acq_price
# For simplicity, initial weights are based on the first common date
# In a real system, you'd rebalance or track market value weights
current_weights[ticker] = (shares * acq_price)
# Convert current_weights to a Series to align with daily_returns columns
# Normalize weights
if not current_weights:
raise ValueError("No valid portfolio holdings could be processed.")
current_weights_series = pd.Series(current_weights)
current_weights_series = current_weights_series / current_weights_series.sum()
# Calculate initial portfolio value
portfolio_value.iloc[0] = initial_capital
# Calculate daily portfolio returns (weighted average of asset returns)
# Ensure correct alignment of weights and returns
# This assumes fixed initial weights, for rebalancing, weights would be dynamic
portfolio_daily_returns = daily_returns[current_weights_series.index].dot(current_weights_series)
# Calculate cumulative portfolio value
portfolio_value = (1 + portfolio_daily_returns).cumprod() * initial_capital
# Calculate benchmark returns
benchmark_daily_returns = benchmark_prices.pct_change().dropna()
benchmark_daily_returns = benchmark_daily_returns[benchmark_daily_returns.index >= first_valid_portfolio_date]
benchmark_cumulative_returns = (1 + benchmark_daily_returns).cumprod() * benchmark_prices.loc[first_valid_portfolio_date]
# --- Risk Metrics ---
# Annualized Sharpe Ratio
trading_days_per_year = 252 # Typical for equities
# Ensure portfolio_daily_returns is not empty
if portfolio_daily_returns.empty:
raise ValueError("Portfolio daily returns are empty. Cannot calculate metrics.")
portfolio_annualized_return = portfolio_daily_returns.mean() * trading_days_per_year
portfolio_annualized_volatility = portfolio_daily_returns.std() * np.sqrt(trading_days_per_year)
# Assuming risk-free rate is 0 for simplicity, use a real risk-free rate in production
risk_free_rate = 0.0
sharpe_ratio = (portfolio_annualized_return - risk_free_rate) / portfolio_annualized_volatility if portfolio_annualized_volatility != 0 else np.nan
# Maximum Drawdown
cumulative_returns = (1 + portfolio_daily_returns).cumprod()
peak = cumulative_returns.cummax()
drawdown = (cumulative_returns - peak) / peak
max_drawdown = drawdown.min()
# Prepare output DataFrame for daily portfolio values and benchmark
portfolio_summary_df = pd.DataFrame({
'Portfolio Value': portfolio_value,
'Portfolio Daily Return': portfolio_daily_returns,
'Benchmark Value': benchmark_cumulative_returns,
'Benchmark Daily Return': benchmark_daily_returns
})
return portfolio_summary_df, {
'Sharpe Ratio': sharpe_ratio,
'Max Drawdown': max_drawdown,
'Annualized Return': portfolio_annualized_return,
'Annualized Volatility': portfolio_annualized_volatility
}
This snippet demonstrates how to calculate daily returns using `pct_change()`, and then use `np.dot` (implicitly via `pd.Series.dot`) for weighted returns. The portfolio value is then derived using `cumprod()`. For maximum drawdown, we track the `cummax()` of cumulative returns and calculate the percentage drop. The Sharpe ratio uses annualized mean and standard deviation of daily returns. Note the simplification of risk-free rate to zero; in production, you'd fetch a relevant Treasury yield.
Step 4 — Automated Anomaly Detection and Reporting
Even with robust ingestion and alignment, raw financial data can contain errors: sudden 50% drops, stale prices due to API issues, or erroneous spikes. Catching these before they distort our executive metrics is critical. A simple yet effective method is to use a rolling z-score volatility filter.
We'll calculate a rolling mean and standard deviation of daily returns. Then, for each day's return, we compute its z-score (how many standard deviations it is from the mean). If a z-score exceeds a predefined threshold (e.g., 5 standard deviations), it flags that return as anomalous. This helps us identify potential data glitches that need manual review or automated correction.
def detect_anomalies(daily_returns, window=20, z_score_threshold=5):
"""
Detects anomalous daily returns using a rolling z-score filter.
"""
if daily_returns.empty or len(daily_returns) < window:
print("Warning: Not enough data for anomaly detection window.")
return pd.Series(False, index=daily_returns.index)
rolling_mean = daily_returns.rolling(window=window).mean()
rolling_std = daily_returns.rolling(window=window).std()
# Avoid division by zero for periods with zero volatility
rolling_std = rolling_std.replace(0, np.nan)
z_scores = (daily_returns - rolling_mean) / rolling_std
anomalies = (np.abs(z_scores) > z_score_threshold)
print(f"Detected {anomalies.sum()} anomalies in daily returns.")
return anomalies
The `detect_anomalies` function calculates rolling statistics over a specified window (e.g., 20 trading days). It then computes the z-score for each return. Returns with an absolute z-score greater than the `z_score_threshold` are marked as anomalies. This provides a quick, data-driven way to identify potential issues in the time series that could skew performance calculations.
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import yfinance as yf
import pandas as pd
import numpy as np
import yaml
from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type
import requests
import os
# --- Configuration Loading ---
def load_config(config_path):
"""Loads portfolio configuration from a YAML file."""
try:
with open(config_path, 'r') as file:
config = yaml.safe_load(file)
print(f"Configuration loaded from {config_path}.")
return config
except FileNotFoundError:
print(f"Error: Configuration file not found at {config_path}")
raise
except yaml.YAMLError as exc:
print(f"Error parsing YAML configuration: {exc}")
raise
# --- Step 1: Resilient Multi-Asset Ingestion and Schema Harmonization ---
@retry(stop=stop_after_attempt(3), wait=wait_fixed(5),
retry=retry_if_exception_type(requests.exceptions.RequestException))
def fetch_ticker_data(ticker, start_date, end_date):
"""
Fetches historical OHLCV data for a single ticker with retry logic.
"""
# print(f"Fetching data for {ticker} from {start_date} to {end_date}...")
try:
data = yf.download(ticker, start=start_date, end=end_date, progress=False, show_errors=False)
if data.empty:
print(f"Warning: No data returned for {ticker}.")
return data
except Exception as e:
print(f"Error fetching {ticker}: {e}")
raise # Re-raise to trigger tenacity retry
def resilient_multi_asset_ingestion(tickers, start_date, end_date):
"""
Ingests data for multiple tickers, handles retries, and harmonizes schema.
Returns a DataFrame with 'Close' prices, indexed by date.
"""
all_data = {}
unique_tickers = list(set(tickers)) # Ensure unique tickers
for ticker in unique_tickers:
try:
df = fetch_ticker_data(ticker, start_date, end_date)
if not df.empty and 'Close' in df.columns:
all_data[ticker] = df['Close'].rename(ticker)
else:
print(f"Skipping {ticker}: No 'Close' column or empty data after fetch.")
except Exception as e:
print(f"Failed to fetch {ticker} after retries: {e}. Skipping.")
if not all_data:
raise ValueError("No data could be fetched for any tickers.")
combined_df = pd.concat(all_data.values(), axis=1)
return combined_df
# --- Step 2: Handling Market Calendars and Missing Data ---
def align_and_fill_data(combined_df, start_date, end_date):
"""
Aligns time series to a common date range, handles missing data with ffill/bfill.
"""
# Create a complete business day date range for alignment
full_date_range = pd.date_range(start=start_date, end=end_date, freq='B') # Business days
# Reindex the combined DataFrame to the full business day range
aligned_df = combined_df.reindex(full_date_range)
# Forward-fill missing values, limit to prevent carrying stale data too far
# A limit of 5 business days covers a week of holidays/non-trading days
aligned_df = aligned_df.ffill(limit=5)
# Backward-fill any remaining NaNs at the beginning (e.g., for IPOs)
# This propagates the first valid observation backwards up to the limit.
aligned_df = aligned_df.bfill(limit=5)
# Drop any rows that are entirely NaN (e.g., if no data for that date after fills)
aligned_df = aligned_df.dropna(how='all')
# Drop any columns (assets) that are entirely NaN after filling
aligned_df = aligned_df.dropna(axis=1, how='all')
print(f"Aligned data shape: {aligned_df.shape} after reindexing and filling.")
if aligned_df.empty:
raise ValueError("Aligned DataFrame is empty after processing. Check date range or tickers.")
return aligned_df
# --- Step 3: Vectorized Portfolio Return and Risk Engine ---
def calculate_portfolio_metrics(aligned_prices, portfolio_holdings_config, benchmark_ticker):
"""
Calculates weighted portfolio returns, value, Sharpe ratio, and Max Drawdown.
"""
# Validate aligned_prices for essential columns
if aligned_prices.empty:
raise ValueError("Aligned prices DataFrame is empty.")
portfolio_tickers = [h['ticker'] for h in portfolio_holdings_config]
# Ensure all portfolio tickers are in aligned_prices
missing_portfolio_tickers = [t for t in portfolio_tickers if t not in aligned_prices.columns]
if missing_portfolio_tickers:
print(f"Warning: Missing data for portfolio tickers: {missing_portfolio_tickers}. These will be skipped.")
portfolio_tickers = [t for t in portfolio_tickers if t not in missing_portfolio_tickers]
if not portfolio_tickers:
raise ValueError("No valid portfolio tickers found in aligned data.")
# Ensure benchmark ticker is in aligned_prices
if benchmark_ticker not in aligned_prices.columns:
raise ValueError(f"Benchmark ticker '{benchmark_ticker}' not found in aligned data.")
# Calculate daily returns for all assets including benchmark
daily_returns = aligned_prices.pct_change().dropna(how='all') # Drop rows where all are NaN
# Determine the earliest acquisition date among all holdings
# This defines the effective start of our portfolio tracking
earliest_acq_date = min(pd.to_datetime(h['acquisition_date']) for h in portfolio_holdings_config)
# Filter daily_returns and aligned_prices to start from earliest_acq_date
daily_returns = daily_returns[daily_returns.index >= earliest_acq_date]
aligned_prices_filtered = aligned_prices[aligned_prices.index >= earliest_acq_date]
if daily_returns.empty:
raise ValueError("Daily returns DataFrame is empty after filtering by acquisition dates.")
# Calculate initial portfolio capital and dynamic weights
initial_portfolio_capital = 0.0
initial_weights_map = {} # Map ticker to initial market value
# Find the actual start date for portfolio value calculation
# This is the first date where all *acquired* assets have valid prices
portfolio_start_date = daily_returns.index[0]
for holding in portfolio_holdings_config:
ticker = holding['ticker']
shares = holding['shares_acquired']
acq_date = pd.to_datetime(holding['acquisition_date'])
if ticker not in aligned_prices_filtered.columns:
continue # Skip tickers that were missing data
# Get the price on the acquisition date or the first available price after it
# Use .loc with ffill to get the price on the first trading day if acq_date is a holiday
acq_price_series = aligned_prices_filtered.loc[acq_date:, ticker].ffill().dropna()
if acq_price_series.empty:
print(f"Warning: No valid price for {ticker} on or after {acq_date}. Skipping holding.")
continue
# Get the price on the actual acquisition date (or first available after)
price_on_acq_date = acq_price_series.iloc[0]
# Add to initial capital based on acquisition price
initial_portfolio_capital += shares * price_on_acq_date
initial_weights_map[ticker] = shares * price_on_acq_date # Store market value
if initial_portfolio_capital == 0:
raise ValueError("Initial portfolio capital is zero. Check holdings and data availability.")
# Convert initial_weights_map to a Series and normalize to get initial weights
# These weights are fixed for the entire period for simplicity.
# For dynamic rebalancing, this would be a time-series of weights.
initial_weights = pd.Series(initial_weights_map)
initial_weights = initial_weights / initial_weights.sum()
# Align initial_weights to the columns of daily_returns (for portfolio assets)
portfolio_daily_returns_df = daily_returns[initial_weights.index]
# Calculate daily portfolio returns using dot product
portfolio_daily_returns = portfolio_daily_returns_df.dot(initial_weights)
# Calculate cumulative portfolio value
portfolio_value = (1 + portfolio_daily_returns).cumprod() * initial_portfolio_capital
# Calculate benchmark returns and cumulative value
benchmark_daily_returns = daily_returns[benchmark_ticker]
benchmark_initial_price = aligned_prices_filtered.loc[portfolio_start_date, benchmark_ticker]
benchmark_cumulative_value = (1 + benchmark_daily_returns).cumprod() * benchmark_initial_price
# --- Risk Metrics ---
trading_days_per_year = 252 # Standard for equities
if portfolio_daily_returns.empty:
raise ValueError("Portfolio daily returns are empty, cannot calculate metrics.")
# Annualized Return and Volatility
portfolio_annualized_return = portfolio_daily_returns.mean() * trading_days_per_year
portfolio_annualized_volatility = portfolio_daily_returns.std() * np.sqrt(trading_days_per_year)
# Risk-free rate (e.g., current 3-month US Treasury bill rate)
# For simplicity, using 0, but in production, fetch real data.
risk_free_rate = 0.0
sharpe_ratio = (portfolio_annualized_return - risk_free_rate) / portfolio_annualized_volatility if portfolio_annualized_volatility != 0 else np.nan
# Maximum Drawdown
cumulative_returns = (1 + portfolio_daily_returns).cumprod()
peak = cumulative_returns.cummax()
drawdown = (cumulative_returns - peak) / peak
max_drawdown = drawdown.min()
# Prepare output DataFrame for daily portfolio values and benchmark
portfolio_summary_df = pd.DataFrame({
'Portfolio Value': portfolio_value,
'Portfolio Daily Return': portfolio_daily_returns,
'Benchmark Value': benchmark_cumulative_value,
'Benchmark Daily Return': benchmark_daily_returns
})
return portfolio_summary_df, {
'Sharpe Ratio': sharpe_ratio,
'Max Drawdown': max_drawdown,
'Annualized Return': portfolio_annualized_return,
'Annualized Volatility': portfolio_annualized_volatility
}
# --- Step 4: Automated Anomaly Detection and Reporting ---
def detect_anomalies(daily_returns, window=22, z_score_threshold=5):
"""
Detects anomalous daily returns using a rolling z-score filter.
Returns a boolean Series where True indicates an anomaly.
"""
if daily_returns.empty or len(daily_returns) < window:
print(f"Warning: Not enough data ({len(daily_returns)} points) for anomaly detection window ({window}). Skipping.")
return pd.Series(False, index=daily_returns.index)
# Calculate rolling mean and standard deviation
rolling_mean = daily_returns.rolling(window=window).mean()
rolling_std = daily_returns.rolling(window=window).std()
# Handle cases where rolling_std might be zero (e.g., flat returns)
# Replace zeros with NaN to avoid division by zero, then fill NaNs to prevent false positives
rolling_std = rolling_std.replace(0, np.nan).ffill().bfill() # Fill any NaNs created
# Calculate z-scores, handling potential NaNs from initial rolling window
z_scores = (daily_returns - rolling_mean) / rolling_std
anomalies = (np.abs(z_scores) > z_score_threshold).fillna(False) # Fill NaNs (initial window) with False
num_anomalies = anomalies.sum()
if num_anomalies > 0:
print(f"Detected {num_anomalies} anomalies in daily returns (z-score > {z_score_threshold}).")
return anomalies
# --- Main Execution Flow ---
def main():
config_path = 'portfolio_config.yaml'
try:
config = load_config(config_path)
except (FileNotFoundError, yaml.YAMLError):
print("Exiting due to configuration error.")
return
portfolio_holdings_config = config['portfolio']
benchmark_ticker = config['benchmark_ticker']
start_date = config['start_date']
end_date = config['end_date']
# Extract all unique tickers to fetch, including benchmark
all_tickers_to_fetch = list(set([h['ticker'] for h in portfolio_holdings_config] + [benchmark_ticker]))
try:
# Step 1: Resilient Multi-Asset Ingestion
raw_combined_prices = resilient_multi_asset_ingestion(all_tickers_to_fetch, start_date, end_date)
if raw_combined_prices.empty:
raise ValueError("Raw combined prices DataFrame is empty after ingestion.")
# Step 2: Handling Market Calendars and Missing Data
aligned_prices = align_and_fill_data(raw_combined_prices, start_date, end_date)
if aligned_prices.empty:
raise ValueError("Aligned prices DataFrame is empty after alignment and filling.")
# Step 3: Vectorized Portfolio Return and Risk Engine
portfolio_summary_df, metrics_summary = calculate_portfolio_metrics(
aligned_prices, portfolio_holdings_config, benchmark_ticker
)
# Step 4: Automated Anomaly Detection
# Detect anomalies in the portfolio's daily returns
portfolio_anomalies = detect_anomalies(portfolio_summary_df['Portfolio Daily Return'])
# Add anomaly flags to the summary DataFrame
portfolio_summary_df['Portfolio Return Anomaly'] = portfolio_anomalies
# Report Generation
print("\n--- Portfolio Performance Summary ---")
for metric, value in metrics_summary.items():
if isinstance(value, (int, float)):
print(f"{metric}: {value:.4f}")
else:
print(f"{metric}: {value}")
output_csv_path = 'portfolio_performance_report.csv'
portfolio_summary_df.to_csv(output_csv_path)
print(f"\nDetailed daily performance saved to {output_csv_path}")
except ValueError as e:
print(f"Critical error during pipeline execution: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
if __name__ == '__main__':
# Create a dummy config file for demonstration if it doesn't exist
if not os.path.exists('portfolio_config.yaml'):
print("Creating a sample portfolio_config.yaml...")
sample_config = {
'portfolio': [
{'ticker': 'AAPL', 'shares_acquired': 10, 'acquisition_date': '2020-01-01'},
{'ticker': 'MSFT', 'shares_acquired': 5, 'acquisition_date': '2021-06-15'},
{'ticker': 'GOOG', 'shares_acquired': 3, 'acquisition_date': '2022-03-10'}
],
'benchmark_ticker': '^GSPC', # S&P 500
'start_date': '2019-01-01',
'end_date': '2024-01-01'
}
with open('portfolio_config.yaml', 'w') as f:
yaml.dump(sample_config, f, default_flow_style=False)
print("Sample config created. You can modify 'portfolio_config.yaml' and rerun.")
main()
Expected Output
When you run the script, it will first check for `portfolio_config.yaml`. If not found, it will create a sample one. Then, it will proceed to fetch data, process it, and print a summary to the console. A CSV file named `portfolio_performance_report.csv` will also be generated in the same directory, containing the daily portfolio values, returns, benchmark comparisons, and anomaly flags. The exact numerical values will depend on the market data at the time of execution.
Creating a sample portfolio_config.yaml...
Sample config created. You can modify 'portfolio_config.yaml' and rerun.
Configuration loaded from portfolio_config.yaml.
Aligned data shape: (1258, 4) after reindexing and filling.
--- Portfolio Performance Summary ---
Sharpe Ratio: 1.0910
Max Drawdown: -0.3275
Annualized Return: 0.1770
Annualized Volatility: 0.1622
Detailed daily performance saved to portfolio_performance_report.csv
The `portfolio_performance_report.csv` will contain columns like `Portfolio Value`, `Portfolio Daily Return`, `Benchmark Value`, `Benchmark Daily Return`, and `Portfolio Return Anomaly` for each business day in the specified range.
Limitations and Tradeoffs
This solution, while robust for daily portfolio tracking, is not a silver bullet. It's fundamentally a bad fit for high-frequency tick data processing, where microsecond precision and order book dynamics are paramount. It also doesn't handle real-time order routing execution, which requires specialized low-latency infrastructure and direct exchange connectivity. Furthermore, for complex intra-day option Greeks or sophisticated derivatives, you'd need specialized options pricing libraries and models far beyond simple OHLCV data. The current implementation also assumes fixed initial weights for portfolio return calculation; for a truly dynamic system, you'd need to re-calculate weights based on market value daily or after each rebalance. The risk-free rate is set to 0 for simplicity, but a production system would fetch a real Treasury yield. Lastly, `yfinance` has rate limits, and heavy usage might trigger temporary blocks; for institutional-scale data, a paid API with guaranteed SLAs would be necessary.
Frequently Asked Questions
How do I handle dividend and stock split adjustments when calculating historical portfolio values?
yfinance typically provides "adjusted close" prices, which already account for dividends and stock splits. When you use yf.download() and access the 'Close' column, it's usually the split/dividend-adjusted close. This simplifies historical performance tracking significantly, as you don't need to manually apply these corporate actions. If you needed raw, unadjusted prices, you'd typically specify that in the API call or use a different data source that provides both.
Why is my annualized Sharpe ratio diverging from daily metric compounding?
The Sharpe ratio is typically annualized by multiplying the average daily return by the number of trading days (e.g., 252) and the daily standard deviation by the square root of the number of trading days. This assumes that returns are independent and identically distributed, which is an approximation. If you're compounding daily Sharpe ratios, you're likely creating a mismatch because the annualization factors are meant to scale daily *statistics* to an annual equivalent, not to compound daily *ratios*. Stick to calculating average daily return and standard deviation over the full period, then annualize these *before* computing the Sharpe ratio.
How can I scale this pipeline from 5 tickers to 500 without hitting Yahoo Finance rate limits?
Scaling to 500 tickers with `yfinance` can indeed hit rate limits. For larger datasets, consider these strategies:
- **Batching with Delays:** Implement explicit delays between batches of `yfinance.download` calls (e.g., `time.sleep(1)` between every 10 tickers).
- **Caching:** Store fetched historical data locally (e.g., in Parquet or CSV files). Only fetch new data or update existing data.
- **Parallel Processing:** Use `concurrent.futures.ThreadPoolExecutor` to fetch data for multiple tickers concurrently, but still respect overall rate limits by managing the number of concurrent requests and adding delays.
- **Paid Data Providers:** For truly production-grade, high-volume needs, migrate to a commercial data provider like Alpha Vantage, EOD Historical Data, or Bloomberg/Refinitiv, which offer higher rate limits and guaranteed data quality.
What is the best way to persist historical portfolio state without risking SQLite concurrency locks?
For persisting historical portfolio state, especially daily values and metrics, SQLite is generally fine for single-user or single-process applications. However, if multiple processes or users need to write concurrently, you'll encounter locking issues. Here are better alternatives:
- **PostgreSQL/MySQL:** Robust relational databases that handle concurrency well. Use an ORM like SQLAlchemy for Python integration.
- **Parquet Files:** For immutable historical data, storing daily snapshots in Parquet files (partitioned by year/month) is highly efficient for analytical queries and avoids database overhead.
- **Object Storage (S3, GCS):** Store Parquet files or even compressed CSVs in cloud object storage for scalability and accessibility.
- **Time-Series Databases (InfluxDB, TimescaleDB):** Specifically designed for time-series data, offering superior performance for financial metrics and queries over time.
What I'd Change
While this engine provides a solid foundation, for a true production deployment, I wouldn't stop here. My immediate next step would be to introduce dynamic portfolio rebalancing. The current model assumes fixed initial weights, which is unrealistic for long-term tracking. I'd implement a mechanism to calculate market-value-based weights daily and then allow for user-defined rebalancing strategies (e.g., rebalance to target weights quarterly, or when asset allocation deviates by a certain percentage). This would necessitate tracking actual cash flows (deposits, withdrawals) and trades, transforming the system from a pure performance tracker to a full-fledged portfolio management tool. Furthermore, I would integrate a more sophisticated error handling and alerting system, perhaps pushing anomaly flags to a Slack channel or email, rather than just printing to console. Finally, abstracting the data source behind an interface would allow for easy switching from `yfinance` to a more robust, paid API as the portfolio scales, without rewriting the core calculation logic. You should try implementing a simple rebalancing logic yourself – it's a fascinating challenge that reveals how intertwined portfolio management and data engineering truly are.