metrics

athena.metrics

Portfolio risk and performance metrics.

Provides functions for calculating Sharpe ratio, Sortino ratio, maximum drawdown, Value at Risk, volatility, win rate, and alpha/beta against market benchmarks. Sub-modules for options-implied analytics (forward curves, theta decay) are available but not re-exported here.

AlphaBetaResult dataclass

Result container for alpha/beta calculations.

Source code in src/athena/metrics/alpha_beta.py

@dataclass
class AlphaBetaResult:
    """Result container for alpha/beta calculations."""
    alpha: float            # Jensen's alpha (daily)
    alpha_annualized: float # Annualized alpha
    beta: float             # Market beta
    r_squared: float        # Coefficient of determination (0-1)
    correlation: float      # Correlation with benchmark

ClosedPosition dataclass

Represents a closed position with its realized gain/loss.

Source code in src/athena/metrics/win_rate.py

@dataclass
class ClosedPosition:
    """Represents a closed position with its realized gain/loss."""
    symbol: str
    quantity: Decimal
    buy_price: Decimal
    sell_price: Decimal
    buy_date: datetime
    sell_date: datetime
    currency: Currency
    realized_gain_loss: Decimal
    realized_gain_loss_percent: Decimal

WinRateResult dataclass

Result of a win rate calculation.

Source code in src/athena/metrics/win_rate.py

@dataclass
class WinRateResult:
    """Result of a win rate calculation."""
    win_rate: float
    total_positions: int
    winning_positions: int
    losing_positions: int
    breakeven_positions: int
    total_gain_loss: Decimal
    average_win: Decimal | None
    average_loss: Decimal | None
    win_loss_ratio: float | None

align_strategy_with_benchmark(strategy_returns, benchmark_returns, rf_manager=None, trading_days_per_year=252)

Align strategy returns with benchmark returns on trading days.

Only includes dates where both strategy and benchmark have data. Optionally includes risk-free rates for each date.

Parameters:

Name	Type	Description	Default
strategy_returns	dict[datetime, float]	Dictionary mapping dates to strategy returns.	required
benchmark_returns	dict[datetime, float]	Dictionary mapping dates to benchmark returns.	required
rf_manager	RiskFreeRateManager | None	Optional risk-free rate manager. If None, uses 0.0 for all dates.	None
trading_days_per_year	int	Number of trading days per year for daily rate calc.	252

Returns:

Type	Description
Tuple[list[float], list[float], list[float], list[datetime]]	Tuple of (aligned_strategy_returns, aligned_benchmark_returns, aligned_rf_rates, aligned_dates).

Raises:

Type	Description
ValueError	If no overlapping dates are found.

Source code in src/athena/metrics/alpha_beta.py

def align_strategy_with_benchmark(
    strategy_returns: dict[datetime, float],
    benchmark_returns: dict[datetime, float],
    rf_manager: RiskFreeRateManager | None = None,
    trading_days_per_year: int = 252
) -> Tuple[list[float], list[float], list[float], list[datetime]]:
    """
    Align strategy returns with benchmark returns on trading days.

    Only includes dates where both strategy and benchmark have data.
    Optionally includes risk-free rates for each date.

    Args:
        strategy_returns: Dictionary mapping dates to strategy returns.
        benchmark_returns: Dictionary mapping dates to benchmark returns.
        rf_manager: Optional risk-free rate manager. If None, uses 0.0 for all dates.
        trading_days_per_year: Number of trading days per year for daily rate calc.

    Returns:
        Tuple of (aligned_strategy_returns, aligned_benchmark_returns,
                  aligned_rf_rates, aligned_dates).

    Raises:
        ValueError: If no overlapping dates are found.
    """
    # Find overlapping dates
    strategy_dates = set(strategy_returns.keys())
    benchmark_dates = set(benchmark_returns.keys())
    common_dates = strategy_dates & benchmark_dates

    if not common_dates:
        raise ValueError(
            "No overlapping dates between strategy and benchmark returns."
        )

    aligned_strategy: list[float] = []
    aligned_benchmark: list[float] = []
    aligned_rf: list[float] = []
    aligned_dates: list[datetime] = []

    for dt in sorted(common_dates):
        dt_date = dt.date()

        # Skip weekends
        if dt_date.weekday() >= 5:
            continue

        # Get risk-free rate for this date
        if rf_manager is not None:
            try:
                annual_rate = rf_manager.get_rate(dt_date)
                daily_rf = float(annual_rate) / trading_days_per_year
            except ValueError:
                # No risk-free rate available, skip this date
                continue
        else:
            daily_rf = 0.0

        aligned_strategy.append(strategy_returns[dt])
        aligned_benchmark.append(benchmark_returns[dt])
        aligned_rf.append(daily_rf)
        aligned_dates.append(dt)

    if len(aligned_strategy) == 0:
        raise ValueError(
            "No overlapping dates after filtering weekends and risk-free rate availability."
        )

    return aligned_strategy, aligned_benchmark, aligned_rf, aligned_dates

calculate_alpha_beta(strategy_returns, benchmark_returns, risk_free_rates=None, trading_days_per_year=252)

Calculate alpha and beta using OLS regression.

Uses the CAPM model: excess_strategy = alpha + beta * excess_benchmark

Parameters:

Name	Type	Description	Default
strategy_returns	list[float]	List of daily strategy returns.	required
benchmark_returns	list[float]	List of daily benchmark returns (same length).	required
risk_free_rates	list[float] | None	Optional list of daily risk-free rates. If None, uses 0.0 (raw returns instead of excess).	None
trading_days_per_year	int	Trading days per year for annualization (default 252).	252

Returns:

Type	Description
AlphaBetaResult	AlphaBetaResult with alpha, beta, r_squared, and correlation.

Raises:

Type	Description
ValueError	If inputs have different lengths or fewer than 2 observations.

Source code in src/athena/metrics/alpha_beta.py

def calculate_alpha_beta(
    strategy_returns: list[float],
    benchmark_returns: list[float],
    risk_free_rates: list[float] | None = None,
    trading_days_per_year: int = 252
) -> AlphaBetaResult:
    """
    Calculate alpha and beta using OLS regression.

    Uses the CAPM model: excess_strategy = alpha + beta * excess_benchmark

    Args:
        strategy_returns: List of daily strategy returns.
        benchmark_returns: List of daily benchmark returns (same length).
        risk_free_rates: Optional list of daily risk-free rates.
                         If None, uses 0.0 (raw returns instead of excess).
        trading_days_per_year: Trading days per year for annualization (default 252).

    Returns:
        AlphaBetaResult with alpha, beta, r_squared, and correlation.

    Raises:
        ValueError: If inputs have different lengths or fewer than 2 observations.
    """
    if len(strategy_returns) != len(benchmark_returns):
        raise ValueError(
            f"Strategy and benchmark returns must have same length. "
            f"Got {len(strategy_returns)} and {len(benchmark_returns)}."
        )

    if len(strategy_returns) < 2:
        raise ValueError("Need at least two return observations.")

    strategy_arr = np.array(strategy_returns, dtype=float)
    benchmark_arr = np.array(benchmark_returns, dtype=float)

    # Calculate excess returns if risk-free rates provided
    if risk_free_rates is not None:
        if len(risk_free_rates) != len(strategy_returns):
            raise ValueError(
                f"Risk-free rates must have same length as returns. "
                f"Got {len(risk_free_rates)} rates and {len(strategy_returns)} returns."
            )
        rf_arr = np.array(risk_free_rates, dtype=float)
        excess_strategy = strategy_arr - rf_arr
        excess_benchmark = benchmark_arr - rf_arr
    else:
        excess_strategy = strategy_arr
        excess_benchmark = benchmark_arr

    # Perform linear regression: excess_strategy = alpha + beta * excess_benchmark
    slope, intercept, r_value, _, _ = stats.linregress(excess_benchmark, excess_strategy)

    beta = float(slope)
    alpha_daily = float(intercept)
    r_squared = float(r_value ** 2)
    correlation = float(r_value)

    # Annualize alpha (compound daily alpha over trading days)
    # For small daily returns, this approximates: alpha_annual ≈ alpha_daily * trading_days
    alpha_annualized = alpha_daily * trading_days_per_year

    return AlphaBetaResult(
        alpha=alpha_daily,
        alpha_annualized=alpha_annualized,
        beta=beta,
        r_squared=r_squared,
        correlation=correlation
    )

calculate_alpha_beta_by_day_cumulative(portfolio, target_currency, benchmark_ticker=BENCHMARK_SP500, rf_manager=None, start_date=None, end_date=None, trading_days_per_year=252, min_observations=20, price_manager=None)

Calculate cumulative alpha/beta for each trading day in a date range.

For each trading day, calculates alpha/beta using all returns from start_date up to that day (cumulative/expanding window).

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
benchmark_ticker	str	Yahoo Finance ticker for benchmark (default: ^GSPC).	BENCHMARK_SP500
rf_manager	RiskFreeRateManager | None	Risk-free rate manager. If None, uses FRED DTB3.	None
start_date	datetime | None	The start date for the calculation (inclusive).	None
end_date	datetime | None	The end date for the calculation (inclusive).	None
trading_days_per_year	int	Trading days per year for annualization (default 252).	252
min_observations	int	Minimum observations before calculating (default 20).	20
price_manager	PricingDataManager | None	PricingDataManager for portfolio valuation.	None

Returns:

Type	Description
dict[datetime, AlphaBetaResult]	Dictionary mapping each trading day to AlphaBetaResult.
dict[datetime, AlphaBetaResult]	Days with insufficient observations are excluded.

Source code in src/athena/metrics/alpha_beta.py

def calculate_alpha_beta_by_day_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    benchmark_ticker: str = BENCHMARK_SP500,
    rf_manager: RiskFreeRateManager | None = None,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    trading_days_per_year: int = 252,
    min_observations: int = 20,
    price_manager: PricingDataManager | None = None
) -> dict[datetime, AlphaBetaResult]:
    """
    Calculate cumulative alpha/beta for each trading day in a date range.

    For each trading day, calculates alpha/beta using all returns from
    start_date up to that day (cumulative/expanding window).

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        benchmark_ticker: Yahoo Finance ticker for benchmark (default: ^GSPC).
        rf_manager: Risk-free rate manager. If None, uses FRED DTB3.
        start_date: The start date for the calculation (inclusive).
        end_date: The end date for the calculation (inclusive).
        trading_days_per_year: Trading days per year for annualization (default 252).
        min_observations: Minimum observations before calculating (default 20).
        price_manager: PricingDataManager for portfolio valuation.

    Returns:
        Dictionary mapping each trading day to AlphaBetaResult.
        Days with insufficient observations are excluded.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate strategy daily returns
    strategy_returns = calculate_daily_returns(portfolio_values)

    if len(strategy_returns) < min_observations:
        return {}

    # Determine date range
    return_dates = sorted(strategy_returns.keys())
    min_dt = return_dates[0].date()
    max_dt = return_dates[-1].date()

    # Fetch benchmark returns
    benchmark_returns = fetch_benchmark_returns(benchmark_ticker, min_dt, max_dt)

    # Create default risk-free manager if not provided
    if rf_manager is None:
        rf_manager = _get_default_risk_free_manager(min_dt, max_dt)

    # Align returns
    aligned_strategy, aligned_benchmark, aligned_rf, aligned_dates = align_strategy_with_benchmark(
        strategy_returns,
        benchmark_returns,
        rf_manager,
        trading_days_per_year
    )

    if len(aligned_strategy) < min_observations:
        return {}

    # Calculate cumulative alpha/beta for each trading day
    result: dict[datetime, AlphaBetaResult] = {}
    cumulative_strategy: list[float] = []
    cumulative_benchmark: list[float] = []
    cumulative_rf: list[float] = []

    for i, dt in enumerate(aligned_dates):
        cumulative_strategy.append(aligned_strategy[i])
        cumulative_benchmark.append(aligned_benchmark[i])
        cumulative_rf.append(aligned_rf[i])

        if len(cumulative_strategy) < min_observations:
            continue

        try:
            ab_result = calculate_alpha_beta(
                cumulative_strategy,
                cumulative_benchmark,
                cumulative_rf,
                trading_days_per_year
            )
            result[dt] = ab_result
        except ValueError:
            # Skip if calculation fails (e.g., zero variance)
            continue

    return result

calculate_alpha_beta_by_day_rolling_window(portfolio, target_currency, window_size, benchmark_ticker=BENCHMARK_SP500, rf_manager=None, start_date=None, end_date=None, trading_days_per_year=252, price_manager=None)

Calculate rolling window alpha/beta for each trading day.

For each trading day, calculates alpha/beta using the trailing window_size trading days of returns.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
window_size	int	Number of trailing trading days to use for each calculation. Common values: 21 (month), 63 (quarter), 252 (year).	required
benchmark_ticker	str	Yahoo Finance ticker for benchmark (default: ^GSPC).	BENCHMARK_SP500
rf_manager	RiskFreeRateManager | None	Risk-free rate manager. If None, uses FRED DTB3.	None
start_date	datetime | None	The start date for the calculation (inclusive).	None
end_date	datetime | None	The end date for the calculation (inclusive).	None
trading_days_per_year	int	Trading days per year for annualization (default 252).	252
price_manager	PricingDataManager | None	PricingDataManager for portfolio valuation.	None

Returns:

Type	Description
dict[datetime, AlphaBetaResult]	Dictionary mapping each trading day to AlphaBetaResult.
dict[datetime, AlphaBetaResult]	Days with insufficient observations are excluded.

Source code in src/athena/metrics/alpha_beta.py

def calculate_alpha_beta_by_day_rolling_window(
    portfolio: Portfolio,
    target_currency: Currency,
    window_size: int,
    benchmark_ticker: str = BENCHMARK_SP500,
    rf_manager: RiskFreeRateManager | None = None,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    trading_days_per_year: int = 252,
    price_manager: PricingDataManager | None = None
) -> dict[datetime, AlphaBetaResult]:
    """
    Calculate rolling window alpha/beta for each trading day.

    For each trading day, calculates alpha/beta using the trailing
    window_size trading days of returns.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        window_size: Number of trailing trading days to use for each calculation.
                     Common values: 21 (month), 63 (quarter), 252 (year).
        benchmark_ticker: Yahoo Finance ticker for benchmark (default: ^GSPC).
        rf_manager: Risk-free rate manager. If None, uses FRED DTB3.
        start_date: The start date for the calculation (inclusive).
        end_date: The end date for the calculation (inclusive).
        trading_days_per_year: Trading days per year for annualization (default 252).
        price_manager: PricingDataManager for portfolio valuation.

    Returns:
        Dictionary mapping each trading day to AlphaBetaResult.
        Days with insufficient observations are excluded.
    """
    if window_size < 2:
        raise ValueError("window_size must be at least 2.")

    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate strategy daily returns
    strategy_returns = calculate_daily_returns(portfolio_values)

    if len(strategy_returns) < window_size:
        return {}

    # Determine date range
    return_dates = sorted(strategy_returns.keys())
    min_dt = return_dates[0].date()
    max_dt = return_dates[-1].date()

    # Fetch benchmark returns
    benchmark_returns = fetch_benchmark_returns(benchmark_ticker, min_dt, max_dt)

    # Create default risk-free manager if not provided
    if rf_manager is None:
        rf_manager = _get_default_risk_free_manager(min_dt, max_dt)

    # Align returns
    aligned_strategy, aligned_benchmark, aligned_rf, aligned_dates = align_strategy_with_benchmark(
        strategy_returns,
        benchmark_returns,
        rf_manager,
        trading_days_per_year
    )

    if len(aligned_strategy) < window_size:
        return {}

    # Calculate rolling alpha/beta for each trading day
    result: dict[datetime, AlphaBetaResult] = {}

    for i in range(window_size - 1, len(aligned_dates)):
        window_strategy = aligned_strategy[i - window_size + 1:i + 1]
        window_benchmark = aligned_benchmark[i - window_size + 1:i + 1]
        window_rf = aligned_rf[i - window_size + 1:i + 1]
        dt = aligned_dates[i]

        try:
            ab_result = calculate_alpha_beta(
                window_strategy,
                window_benchmark,
                window_rf,
                trading_days_per_year
            )
            result[dt] = ab_result
        except ValueError:
            # Skip if calculation fails
            continue

    return result

calculate_alpha_beta_cumulative(portfolio, target_currency, benchmark_ticker=BENCHMARK_SP500, rf_manager=None, start_date=None, end_date=None, trading_days_per_year=252, price_manager=None)

Calculate cumulative alpha and beta for a portfolio vs benchmark.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
benchmark_ticker	str	Yahoo Finance ticker for benchmark (default: ^GSPC).	BENCHMARK_SP500
rf_manager	RiskFreeRateManager | None	Risk-free rate manager. If None, uses FRED DTB3.	None
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
trading_days_per_year	int	Trading days per year for annualization (default 252).	252
price_manager	PricingDataManager | None	PricingDataManager for portfolio valuation. If None, uses YFinancePricingDataManager.	None

Returns:

Type	Description
AlphaBetaResult	AlphaBetaResult with alpha, beta, r_squared, and correlation.

Raises:

Type	Description
ValueError	If insufficient data or calculation error.

Source code in src/athena/metrics/alpha_beta.py

def calculate_alpha_beta_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    benchmark_ticker: str = BENCHMARK_SP500,
    rf_manager: RiskFreeRateManager | None = None,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    trading_days_per_year: int = 252,
    price_manager: PricingDataManager | None = None
) -> AlphaBetaResult:
    """
    Calculate cumulative alpha and beta for a portfolio vs benchmark.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        benchmark_ticker: Yahoo Finance ticker for benchmark (default: ^GSPC).
        rf_manager: Risk-free rate manager. If None, uses FRED DTB3.
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        trading_days_per_year: Trading days per year for annualization (default 252).
        price_manager: PricingDataManager for portfolio valuation.
                       If None, uses YFinancePricingDataManager.

    Returns:
        AlphaBetaResult with alpha, beta, r_squared, and correlation.

    Raises:
        ValueError: If insufficient data or calculation error.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        raise ValueError("Need at least two portfolio value observations.")

    # Calculate strategy daily returns
    strategy_returns = calculate_daily_returns(portfolio_values)

    if len(strategy_returns) < 2:
        raise ValueError("Need at least two return observations.")

    # Determine date range
    return_dates = sorted(strategy_returns.keys())
    min_date = return_dates[0].date()
    max_date = return_dates[-1].date()

    # Fetch benchmark returns
    benchmark_returns = fetch_benchmark_returns(
        benchmark_ticker,
        min_date,
        max_date
    )

    # Create default risk-free manager if not provided
    if rf_manager is None:
        rf_manager = _get_default_risk_free_manager(min_date, max_date)

    # Align returns
    aligned_strategy, aligned_benchmark, aligned_rf, _ = align_strategy_with_benchmark(
        strategy_returns,
        benchmark_returns,
        rf_manager,
        trading_days_per_year
    )

    if len(aligned_strategy) < 2:
        raise ValueError(
            "Need at least two aligned observations after filtering to trading days."
        )

    return calculate_alpha_beta(
        aligned_strategy,
        aligned_benchmark,
        aligned_rf,
        trading_days_per_year
    )

calculate_alpha_beta_from_values(portfolio_values, benchmark_ticker=BENCHMARK_SP500, rf_manager=None, trading_days_per_year=252)

Calculate alpha/beta directly from portfolio values dictionary.

Convenience function when you already have portfolio values and don't need to calculate them from a Portfolio object.

Parameters:

Name	Type	Description	Default
portfolio_values	dict[datetime, Decimal]	Dictionary mapping dates to portfolio values.	required
benchmark_ticker	str	Yahoo Finance ticker for benchmark (default: ^GSPC).	BENCHMARK_SP500
rf_manager	RiskFreeRateManager | None	Risk-free rate manager. If None, uses FRED DTB3.	None
trading_days_per_year	int	Trading days per year for annualization (default 252).	252

Returns:

Type	Description
AlphaBetaResult	AlphaBetaResult with alpha, beta, r_squared, and correlation.

Raises:

Type	Description
ValueError	If insufficient data or calculation error.

Source code in src/athena/metrics/alpha_beta.py

def calculate_alpha_beta_from_values(
    portfolio_values: dict[datetime, Decimal],
    benchmark_ticker: str = BENCHMARK_SP500,
    rf_manager: RiskFreeRateManager | None = None,
    trading_days_per_year: int = 252
) -> AlphaBetaResult:
    """
    Calculate alpha/beta directly from portfolio values dictionary.

    Convenience function when you already have portfolio values and don't
    need to calculate them from a Portfolio object.

    Args:
        portfolio_values: Dictionary mapping dates to portfolio values.
        benchmark_ticker: Yahoo Finance ticker for benchmark (default: ^GSPC).
        rf_manager: Risk-free rate manager. If None, uses FRED DTB3.
        trading_days_per_year: Trading days per year for annualization (default 252).

    Returns:
        AlphaBetaResult with alpha, beta, r_squared, and correlation.

    Raises:
        ValueError: If insufficient data or calculation error.
    """
    # Calculate strategy daily returns
    strategy_returns = calculate_daily_returns(portfolio_values)

    if len(strategy_returns) < 2:
        raise ValueError("Need at least two return observations.")

    # Determine date range
    return_dates = sorted(strategy_returns.keys())
    min_date = return_dates[0].date()
    max_date = return_dates[-1].date()

    # Fetch benchmark returns
    benchmark_returns = fetch_benchmark_returns(benchmark_ticker, min_date, max_date)

    # Create default risk-free manager if not provided
    if rf_manager is None:
        rf_manager = _get_default_risk_free_manager(min_date, max_date)

    # Align returns
    aligned_strategy, aligned_benchmark, aligned_rf, _ = align_strategy_with_benchmark(
        strategy_returns,
        benchmark_returns,
        rf_manager,
        trading_days_per_year
    )

    if len(aligned_strategy) < 2:
        raise ValueError(
            "Need at least two aligned observations after filtering to trading days."
        )

    return calculate_alpha_beta(
        aligned_strategy,
        aligned_benchmark,
        aligned_rf,
        trading_days_per_year
    )

fetch_benchmark_returns(benchmark_ticker, min_date, max_date, force_cache_refresh=False)

Fetch benchmark price data and calculate daily returns.

Parameters:

Name	Type	Description	Default
benchmark_ticker	str	Yahoo Finance ticker (e.g., "^GSPC" for S&P 500).	required
min_date	date	Start date for data.	required
max_date	date	End date for data.	required
force_cache_refresh	bool	Whether to force refresh cached data.	False

Returns:

Type	Description
dict[datetime, float]	Dictionary mapping dates to daily returns.

Raises:

Type	Description
ValueError	If no benchmark data is available.

Source code in src/athena/metrics/alpha_beta.py

def fetch_benchmark_returns(
    benchmark_ticker: str,
    min_date: date,
    max_date: date,
    force_cache_refresh: bool = False
) -> dict[datetime, float]:
    """
    Fetch benchmark price data and calculate daily returns.

    Args:
        benchmark_ticker: Yahoo Finance ticker (e.g., "^GSPC" for S&P 500).
        min_date: Start date for data.
        max_date: End date for data.
        force_cache_refresh: Whether to force refresh cached data.

    Returns:
        Dictionary mapping dates to daily returns.

    Raises:
        ValueError: If no benchmark data is available.
    """
    df = fetch_yfinance_data(benchmark_ticker, min_date, max_date, force_cache_refresh)

    if df.empty:
        raise ValueError(f"No benchmark data available for {benchmark_ticker}")

    # Calculate daily returns from Close prices
    returns: dict[datetime, float] = {}
    sorted_df = df.sort_values('Date')

    prev_close = None
    for _, row in sorted_df.iterrows():
        current_close = float(row['Close'])
        current_date = row['Date']

        if prev_close is not None and prev_close != 0:
            daily_return = (current_close - prev_close) / prev_close
            # Convert date to datetime for consistency with portfolio returns
            dt = datetime.combine(current_date, datetime.min.time())
            returns[dt] = daily_return

        prev_close = current_close

    return returns

calculate_drawdown_series(portfolio_values)

Calculate the drawdown at each point in time from a time series of portfolio values.

Parameters:

Name	Type	Description	Default
portfolio_values	dict[datetime, Decimal]	Dictionary mapping dates to portfolio values.	required

Returns:

Type	Description
dict[datetime, float]	Dictionary mapping dates to drawdown values (negative floats).
dict[datetime, float]	A drawdown of -0.10 means the portfolio is 10% below its peak.

Source code in src/athena/metrics/max_drawdown.py

def calculate_drawdown_series(
    portfolio_values: dict[datetime, Decimal]
) -> dict[datetime, float]:
    """
    Calculate the drawdown at each point in time from a time series of portfolio values.

    Args:
        portfolio_values: Dictionary mapping dates to portfolio values.

    Returns:
        Dictionary mapping dates to drawdown values (negative floats).
        A drawdown of -0.10 means the portfolio is 10% below its peak.
    """
    if len(portfolio_values) < 1:
        return {}

    sorted_dates = sorted(portfolio_values.keys())
    values = [float(portfolio_values[d]) for d in sorted_dates]

    drawdowns: dict[datetime, float] = {}
    peak_value = values[0]

    for date, value in zip(sorted_dates, values):
        if value > peak_value:
            peak_value = value

        if peak_value > 0:
            drawdowns[date] = (value - peak_value) / peak_value
        else:
            drawdowns[date] = 0.0

    return drawdowns

calculate_max_drawdown(portfolio_values)

Calculate the maximum drawdown from a time series of portfolio values.

Maximum drawdown measures the largest peak-to-trough decline in portfolio value, expressed as a percentage of the peak value.

Parameters:

Name	Type	Description	Default
portfolio_values	dict[datetime, Decimal]	Dictionary mapping dates to portfolio values.	required

Returns:

Type	Description
float	Tuple of (max_drawdown, peak_date, trough_date).
datetime | None	max_drawdown is a negative float (e.g., -0.20 = 20% drawdown).
datetime | None	peak_date and trough_date are None if insufficient data.

Raises:

Type	Description
ValueError	If fewer than two observations.

Source code in src/athena/metrics/max_drawdown.py

def calculate_max_drawdown(
    portfolio_values: dict[datetime, Decimal]
) -> Tuple[float, datetime | None, datetime | None]:
    """
    Calculate the maximum drawdown from a time series of portfolio values.

    Maximum drawdown measures the largest peak-to-trough decline in portfolio
    value, expressed as a percentage of the peak value.

    Args:
        portfolio_values: Dictionary mapping dates to portfolio values.

    Returns:
        Tuple of (max_drawdown, peak_date, trough_date).
        max_drawdown is a negative float (e.g., -0.20 = 20% drawdown).
        peak_date and trough_date are None if insufficient data.

    Raises:
        ValueError: If fewer than two observations.
    """
    if len(portfolio_values) < 2:
        raise ValueError("Need at least two portfolio value observations.")

    sorted_dates = sorted(portfolio_values.keys())
    values = [float(portfolio_values[d]) for d in sorted_dates]

    max_drawdown = 0.0
    peak_value = values[0]
    peak_date: datetime | None = sorted_dates[0]
    trough_date: datetime | None = None
    current_peak_date = sorted_dates[0]

    for i, (date, value) in enumerate(zip(sorted_dates, values)):
        if value > peak_value:
            peak_value = value
            current_peak_date = date

        if peak_value > 0:
            drawdown = (value - peak_value) / peak_value
            if drawdown < max_drawdown:
                max_drawdown = drawdown
                peak_date = current_peak_date
                trough_date = date

    return max_drawdown, peak_date, trough_date

calculate_max_drawdown_by_day_cumulative(portfolio, target_currency, start_date=None, end_date=None, min_observations=2)

Calculate cumulative maximum drawdown for each day in a date range.

For each day, calculates the maximum drawdown using all portfolio values from start_date up to that day (cumulative/expanding window).

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
min_observations	int	Minimum number of observations required before calculating max drawdown. Defaults to 2.	2

Returns:

Type	Description
dict[datetime, float]	Dictionary mapping each date to the cumulative max drawdown up to that date.
dict[datetime, float]	Days with insufficient observations are excluded.

Source code in src/athena/metrics/max_drawdown.py

def calculate_max_drawdown_by_day_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    min_observations: int = 2
) -> dict[datetime, float]:
    """
    Calculate cumulative maximum drawdown for each day in a date range.

    For each day, calculates the maximum drawdown using all portfolio values
    from start_date up to that day (cumulative/expanding window).

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        min_observations: Minimum number of observations required
                          before calculating max drawdown. Defaults to 2.

    Returns:
        Dictionary mapping each date to the cumulative max drawdown up to that date.
        Days with insufficient observations are excluded.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < min_observations:
        return {}

    sorted_dates = sorted(portfolio_values.keys())
    values = [float(portfolio_values[d]) for d in sorted_dates]

    result: dict[datetime, float] = {}
    max_drawdown = 0.0
    peak_value = values[0]

    for i, (date, value) in enumerate(zip(sorted_dates, values)):
        if value > peak_value:
            peak_value = value

        if peak_value > 0:
            drawdown = (value - peak_value) / peak_value
            if drawdown < max_drawdown:
                max_drawdown = drawdown

        if i + 1 >= min_observations:
            result[date] = max_drawdown

    return result

calculate_max_drawdown_by_day_rolling_window(portfolio, target_currency, window_size, start_date=None, end_date=None)

Calculate rolling window maximum drawdown for each day in a date range.

For each day, calculates the maximum drawdown using the trailing window_size days of portfolio values.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
window_size	int	Number of trailing days to use for each calculation. Common values: 30, 60, 90, 252 (trading year).	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None

Returns:

Type	Description
dict[datetime, float]	Dictionary mapping each date to the rolling max drawdown.
dict[datetime, float]	Days with insufficient observations (fewer than window_size) are excluded.

Source code in src/athena/metrics/max_drawdown.py

def calculate_max_drawdown_by_day_rolling_window(
    portfolio: Portfolio,
    target_currency: Currency,
    window_size: int,
    start_date: datetime | None = None,
    end_date: datetime | None = None
) -> dict[datetime, float]:
    """
    Calculate rolling window maximum drawdown for each day in a date range.

    For each day, calculates the maximum drawdown using the trailing
    window_size days of portfolio values.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        window_size: Number of trailing days to use for each calculation.
                     Common values: 30, 60, 90, 252 (trading year).
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.

    Returns:
        Dictionary mapping each date to the rolling max drawdown.
        Days with insufficient observations (fewer than window_size) are excluded.
    """
    if window_size < 2:
        raise ValueError("window_size must be at least 2.")

    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < window_size:
        return {}

    sorted_dates = sorted(portfolio_values.keys())
    values = [float(portfolio_values[d]) for d in sorted_dates]

    result: dict[datetime, float] = {}

    for i in range(window_size - 1, len(sorted_dates)):
        window_values = values[i - window_size + 1:i + 1]
        date = sorted_dates[i]

        # Calculate max drawdown for this window
        max_drawdown = 0.0
        peak_value = window_values[0]

        for value in window_values:
            if value > peak_value:
                peak_value = value

            if peak_value > 0:
                drawdown = (value - peak_value) / peak_value
                if drawdown < max_drawdown:
                    max_drawdown = drawdown

        result[date] = max_drawdown

    return result

calculate_max_drawdown_cumulative(portfolio, target_currency, start_date=None, end_date=None)

Calculate the maximum drawdown for a portfolio over a date range.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None

Returns:

Type	Description
float	Tuple of (max_drawdown, peak_date, trough_date).
datetime | None	max_drawdown is a negative float (e.g., -0.20 = 20% drawdown).

Raises:

Type	Description
ValueError	If fewer than two observations.

Source code in src/athena/metrics/max_drawdown.py

def calculate_max_drawdown_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    start_date: datetime | None = None,
    end_date: datetime | None = None
) -> Tuple[float, datetime | None, datetime | None]:
    """
    Calculate the maximum drawdown for a portfolio over a date range.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.

    Returns:
        Tuple of (max_drawdown, peak_date, trough_date).
        max_drawdown is a negative float (e.g., -0.20 = 20% drawdown).

    Raises:
        ValueError: If fewer than two observations.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        raise ValueError("Need at least two portfolio value observations.")

    return calculate_max_drawdown(portfolio_values)

calculate_max_drawdown_from_values(portfolio_values)

Calculate maximum drawdown directly from a dictionary of portfolio values.

This is a convenience function when you already have portfolio values and don't need to calculate them from a Portfolio object.

Parameters:

Name	Type	Description	Default
portfolio_values	dict[datetime, Decimal]	Dictionary mapping dates to portfolio values.	required

Returns:

Type	Description
float	Tuple of (max_drawdown, peak_date, trough_date).
datetime | None	max_drawdown is a negative float (e.g., -0.20 = 20% drawdown).

Raises:

Type	Description
ValueError	If fewer than two observations.

Source code in src/athena/metrics/max_drawdown.py

def calculate_max_drawdown_from_values(
    portfolio_values: dict[datetime, Decimal]
) -> Tuple[float, datetime | None, datetime | None]:
    """
    Calculate maximum drawdown directly from a dictionary of portfolio values.

    This is a convenience function when you already have portfolio values
    and don't need to calculate them from a Portfolio object.

    Args:
        portfolio_values: Dictionary mapping dates to portfolio values.

    Returns:
        Tuple of (max_drawdown, peak_date, trough_date).
        max_drawdown is a negative float (e.g., -0.20 = 20% drawdown).

    Raises:
        ValueError: If fewer than two observations.
    """
    return calculate_max_drawdown(portfolio_values)

calculate_daily_returns(portfolio_values)

Calculate daily returns from a time series of portfolio values.

Parameters:

Name	Type	Description	Default
portfolio_values	dict[datetime, Decimal]	Dictionary mapping dates to portfolio values.	required

Returns:

Type	Description
dict[datetime, float]	Dictionary mapping dates to daily returns (as floats).
dict[datetime, float]	The first date will not have a return (needs previous day).

Source code in src/athena/metrics/sharpe.py

def calculate_daily_returns(
    portfolio_values: dict[datetime, Decimal]
) -> dict[datetime, float]:
    """
    Calculate daily returns from a time series of portfolio values.

    Args:
        portfolio_values: Dictionary mapping dates to portfolio values.

    Returns:
        Dictionary mapping dates to daily returns (as floats).
        The first date will not have a return (needs previous day).
    """
    if len(portfolio_values) < 2:
        return {}

    sorted_dates = sorted(portfolio_values.keys())
    returns: dict[datetime, float] = {}

    for i in range(1, len(sorted_dates)):
        prev_date = sorted_dates[i - 1]
        curr_date = sorted_dates[i]

        prev_value = float(portfolio_values[prev_date])
        curr_value = float(portfolio_values[curr_date])

        if prev_value != 0:
            daily_return = (curr_value - prev_value) / prev_value
            returns[curr_date] = daily_return

    return returns

calculate_sharpe_ratio(returns, annual_risk_free_rate, periods_in_year=365)

Calculate daily and annualized Sharpe ratios from a list of returns.

Parameters:

Name	Type	Description	Default
returns	list[float]	List of daily returns as decimals (e.g., 0.002 = 0.2%).	required
annual_risk_free_rate	float	Annual nominal risk-free rate (e.g., 0.03 for 3%).	required
periods_in_year	int	Trading periods in a year (365 for daily).	365

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_sharpe, annual_sharpe).

Raises:

Type	Description
ValueError	If fewer than two observations or zero standard deviation.

Source code in src/athena/metrics/sharpe.py

def calculate_sharpe_ratio(
    returns: list[float],
    annual_risk_free_rate: float,
    periods_in_year: int = 365
) -> Tuple[float, float]:
    """
    Calculate daily and annualized Sharpe ratios from a list of returns.

    Args:
        returns: List of daily returns as decimals (e.g., 0.002 = 0.2%).
        annual_risk_free_rate: Annual nominal risk-free rate (e.g., 0.03 for 3%).
        periods_in_year: Trading periods in a year (365 for daily).

    Returns:
        Tuple of (daily_sharpe, annual_sharpe).

    Raises:
        ValueError: If fewer than two observations or zero standard deviation.
    """
    if len(returns) < 2:
        raise ValueError("Need at least two return observations.")

    daily_rf = annual_risk_free_rate / periods_in_year

    excess_returns = np.array(returns, dtype=float) - daily_rf
    mean_excess = excess_returns.mean()
    std_excess = excess_returns.std(ddof=1)

    if np.isclose(std_excess, 0.0):
        raise ValueError("Sample standard deviation is zero; Sharpe is undefined.")

    daily_sharpe = float(mean_excess / std_excess)
    annual_sharpe = daily_sharpe * np.sqrt(periods_in_year)

    return daily_sharpe, annual_sharpe

calculate_sharpe_ratio_by_day_cumulative(portfolio, target_currency, annual_risk_free_rate, start_date=None, end_date=None, periods_in_year=365, min_observations=2)

Calculate cumulative Sharpe ratios for each day in a date range.

For each day, calculates the Sharpe ratio using all returns from start_date up to that day (cumulative/expanding window).

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
annual_risk_free_rate	float	Annual nominal risk-free rate (e.g., 0.03 for 3%).	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
periods_in_year	int	Trading periods in a year (365 for daily, 252 for US equities trading days).	365
min_observations	int	Minimum number of return observations required before calculating Sharpe. Defaults to 2.	2

Returns:

Type	Description
dict[datetime, Tuple[float, float]]	Dictionary mapping each date to a tuple of (daily_sharpe, annual_sharpe).
dict[datetime, Tuple[float, float]]	Days with insufficient observations are excluded.

Source code in src/athena/metrics/sharpe.py

def calculate_sharpe_ratio_by_day_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    annual_risk_free_rate: float,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    periods_in_year: int = 365,
    min_observations: int = 2
) -> dict[datetime, Tuple[float, float]]:
    """
    Calculate cumulative Sharpe ratios for each day in a date range.

    For each day, calculates the Sharpe ratio using all returns from
    start_date up to that day (cumulative/expanding window).

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        annual_risk_free_rate: Annual nominal risk-free rate (e.g., 0.03 for 3%).
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        periods_in_year: Trading periods in a year (365 for daily, 252 for
                         US equities trading days).
        min_observations: Minimum number of return observations required
                          before calculating Sharpe. Defaults to 2.

    Returns:
        Dictionary mapping each date to a tuple of (daily_sharpe, annual_sharpe).
        Days with insufficient observations are excluded.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < min_observations:
        return {}

    # Calculate cumulative Sharpe for each day
    sorted_dates = sorted(daily_returns.keys())
    daily_rf = annual_risk_free_rate / periods_in_year

    result: dict[datetime, Tuple[float, float]] = {}
    cumulative_returns: list[float] = []

    for date in sorted_dates:
        cumulative_returns.append(daily_returns[date])

        if len(cumulative_returns) < min_observations:
            continue

        excess_returns = np.array(cumulative_returns, dtype=float) - daily_rf
        mean_excess = excess_returns.mean()
        std_excess = excess_returns.std(ddof=1)

        if np.isclose(std_excess, 0.0):
            # Skip days where std is zero
            continue

        daily_sharpe = float(mean_excess / std_excess)
        annual_sharpe = daily_sharpe * np.sqrt(periods_in_year)

        result[date] = (daily_sharpe, annual_sharpe)

    return result

calculate_sharpe_ratio_by_day_rolling_window(portfolio, target_currency, annual_risk_free_rate, window_size, start_date=None, end_date=None, periods_in_year=365)

Calculate rolling window Sharpe ratios for each day in a date range.

For each day, calculates the Sharpe ratio using the trailing window_size days of returns.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
annual_risk_free_rate	float	Annual nominal risk-free rate (e.g., 0.03 for 3%).	required
window_size	int	Number of trailing days to use for each calculation. Common values: 30, 60, 90, 252 (trading year).	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
periods_in_year	int	Trading periods in a year (365 for daily, 252 for US equities trading days).	365

Returns:

Type	Description
dict[datetime, Tuple[float, float]]	Dictionary mapping each date to a tuple of (daily_sharpe, annual_sharpe).
dict[datetime, Tuple[float, float]]	Days with insufficient observations (fewer than window_size) are excluded.

Source code in src/athena/metrics/sharpe.py

def calculate_sharpe_ratio_by_day_rolling_window(
    portfolio: Portfolio,
    target_currency: Currency,
    annual_risk_free_rate: float,
    window_size: int,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    periods_in_year: int = 365
) -> dict[datetime, Tuple[float, float]]:
    """
    Calculate rolling window Sharpe ratios for each day in a date range.

    For each day, calculates the Sharpe ratio using the trailing
    window_size days of returns.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        annual_risk_free_rate: Annual nominal risk-free rate (e.g., 0.03 for 3%).
        window_size: Number of trailing days to use for each calculation.
                     Common values: 30, 60, 90, 252 (trading year).
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        periods_in_year: Trading periods in a year (365 for daily, 252 for
                         US equities trading days).

    Returns:
        Dictionary mapping each date to a tuple of (daily_sharpe, annual_sharpe).
        Days with insufficient observations (fewer than window_size) are excluded.
    """
    if window_size < 2:
        raise ValueError("window_size must be at least 2.")

    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < window_size:
        return {}

    # Calculate rolling Sharpe for each day
    sorted_dates = sorted(daily_returns.keys())
    returns_list = [daily_returns[d] for d in sorted_dates]
    daily_rf = annual_risk_free_rate / periods_in_year

    result: dict[datetime, Tuple[float, float]] = {}

    for i in range(window_size - 1, len(sorted_dates)):
        window_returns = returns_list[i - window_size + 1:i + 1]
        date = sorted_dates[i]

        excess_returns = np.array(window_returns, dtype=float) - daily_rf
        mean_excess = excess_returns.mean()
        std_excess = excess_returns.std(ddof=1)

        if np.isclose(std_excess, 0.0):
            # Skip days where std is zero
            continue

        daily_sharpe = float(mean_excess / std_excess)
        annual_sharpe = daily_sharpe * np.sqrt(periods_in_year)

        result[date] = (daily_sharpe, annual_sharpe)

    return result

calculate_sharpe_ratio_cumulative(portfolio, target_currency, annual_risk_free_rate, start_date=None, end_date=None, periods_in_year=365)

Calculate the cumulative Sharpe ratio for a portfolio over a date range.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
annual_risk_free_rate	float	Annual nominal risk-free rate (e.g., 0.03 for 3%).	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
periods_in_year	int	Trading periods in a year (365 for daily, 252 for US equities trading days).	365

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_sharpe, annual_sharpe).

Raises:

Type	Description
ValueError	If fewer than two observations or zero standard deviation.

Source code in src/athena/metrics/sharpe.py

def calculate_sharpe_ratio_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    annual_risk_free_rate: float,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    periods_in_year: int = 365
) -> Tuple[float, float]:
    """
    Calculate the cumulative Sharpe ratio for a portfolio over a date range.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        annual_risk_free_rate: Annual nominal risk-free rate (e.g., 0.03 for 3%).
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        periods_in_year: Trading periods in a year (365 for daily, 252 for
                         US equities trading days).

    Returns:
        Tuple of (daily_sharpe, annual_sharpe).

    Raises:
        ValueError: If fewer than two observations or zero standard deviation.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        raise ValueError("Need at least two portfolio value observations.")

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < 2:
        raise ValueError("Need at least two return observations.")

    # Calculate Sharpe ratio
    returns_list = [daily_returns[d] for d in sorted(daily_returns.keys())]
    return calculate_sharpe_ratio(returns_list, annual_risk_free_rate, periods_in_year)

calculate_sharpe_ratio_from_values(portfolio_values, annual_risk_free_rate, periods_in_year=365)

Calculate Sharpe ratios directly from a dictionary of portfolio values.

This is a convenience function when you already have portfolio values and don't need to calculate them from a Portfolio object.

Parameters:

Name	Type	Description	Default
portfolio_values	dict[datetime, Decimal]	Dictionary mapping dates to portfolio values.	required
annual_risk_free_rate	float	Annual nominal risk-free rate (e.g., 0.03 for 3%).	required
periods_in_year	int	Trading periods in a year (365 for daily).	365

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_sharpe, annual_sharpe).

Raises:

Type	Description
ValueError	If fewer than two observations or zero standard deviation.

Source code in src/athena/metrics/sharpe.py

def calculate_sharpe_ratio_from_values(
    portfolio_values: dict[datetime, Decimal],
    annual_risk_free_rate: float,
    periods_in_year: int = 365
) -> Tuple[float, float]:
    """
    Calculate Sharpe ratios directly from a dictionary of portfolio values.

    This is a convenience function when you already have portfolio values
    and don't need to calculate them from a Portfolio object.

    Args:
        portfolio_values: Dictionary mapping dates to portfolio values.
        annual_risk_free_rate: Annual nominal risk-free rate (e.g., 0.03 for 3%).
        periods_in_year: Trading periods in a year (365 for daily).

    Returns:
        Tuple of (daily_sharpe, annual_sharpe).

    Raises:
        ValueError: If fewer than two observations or zero standard deviation.
    """
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < 2:
        raise ValueError("Need at least two return observations.")

    returns_list = [daily_returns[d] for d in sorted(daily_returns.keys())]

    return calculate_sharpe_ratio(returns_list, annual_risk_free_rate, periods_in_year)

align_returns_with_risk_free_rates(portfolio_returns, rf_manager, trading_days_per_year=252)

Align portfolio returns with risk-free rates on trading days only.

Includes dates where portfolio returns exist and either: - Actual risk-free rate data is available for that date, OR - It's a recent weekday where FRED data may not be published yet (uses most recent available rate as fallback)

Weekends are always excluded.

Parameters:

Name	Type	Description	Default
portfolio_returns	dict[datetime, float]	Dictionary mapping dates to portfolio returns.	required
rf_manager	RiskFreeRateManager	Risk-free rate manager to get daily rates from.	required
trading_days_per_year	int	Number of trading days per year for daily rate calc.	252

Returns:

Type	Description
list[float]	Tuple of (aligned_portfolio_returns, aligned_rf_rates, aligned_dates).
list[float]	All three lists have the same length and correspond to each other.

Raises:

Type	Description
ValueError	If no overlapping dates are found.

Source code in src/athena/metrics/sharpe_advanced.py

def align_returns_with_risk_free_rates(
    portfolio_returns: dict[datetime, float],
    rf_manager: RiskFreeRateManager,
    trading_days_per_year: int = 252
) -> Tuple[list[float], list[float], list[datetime]]:
    """
    Align portfolio returns with risk-free rates on trading days only.

    Includes dates where portfolio returns exist and either:
    - Actual risk-free rate data is available for that date, OR
    - It's a recent weekday where FRED data may not be published yet
      (uses most recent available rate as fallback)

    Weekends are always excluded.

    Args:
        portfolio_returns: Dictionary mapping dates to portfolio returns.
        rf_manager: Risk-free rate manager to get daily rates from.
        trading_days_per_year: Number of trading days per year for daily rate calc.

    Returns:
        Tuple of (aligned_portfolio_returns, aligned_rf_rates, aligned_dates).
        All three lists have the same length and correspond to each other.

    Raises:
        ValueError: If no overlapping dates are found.
    """
    aligned_returns: list[float] = []
    aligned_rf_rates: list[float] = []
    aligned_dates: list[datetime] = []

    sorted_dates = sorted(portfolio_returns.keys())

    # Get the set of actual trading days from the risk-free rate manager
    if hasattr(rf_manager, 'get_rates_for_range') and sorted_dates:
        min_date = sorted_dates[0].date()
        max_date = sorted_dates[-1].date()
        available_rates = rf_manager.get_rates_for_range(min_date, max_date)
        trading_days = set(available_rates.keys())
        # Track the most recent available rate for fallback on recent days
        last_known_rate: float | None = None
        if available_rates:
            most_recent_date = max(available_rates.keys())
            last_known_rate = float(available_rates[most_recent_date])
    else:
        # Fallback for managers that don't support get_rates_for_range
        trading_days = None
        last_known_rate = None

    for dt in sorted_dates:
        dt_date = dt.date()

        # Always skip weekends (Saturday=5, Sunday=6)
        if dt_date.weekday() >= 5:
            continue

        # Check if we have actual FRED data for this date
        has_fred_data = trading_days is None or dt_date in trading_days

        if has_fred_data:
            try:
                annual_rate = rf_manager.get_rate(dt_date)
                daily_rf = float(annual_rate) / trading_days_per_year
                # Update last known rate for potential future fallback
                last_known_rate = float(annual_rate)

                aligned_returns.append(portfolio_returns[dt])
                aligned_rf_rates.append(daily_rf)
                aligned_dates.append(dt)
            except ValueError:
                # No risk-free rate available for this date, skip it
                continue
        elif last_known_rate is not None:
            # Recent weekday without FRED data - use last known rate as fallback
            # This handles cases where FRED data lags by a day or two
            daily_rf = last_known_rate / trading_days_per_year

            aligned_returns.append(portfolio_returns[dt])
            aligned_rf_rates.append(daily_rf)
            aligned_dates.append(dt)
        # else: skip - no fallback rate available

    if len(aligned_returns) == 0:
        raise ValueError(
            "No overlapping dates between portfolio returns and risk-free rate data."
        )

    return aligned_returns, aligned_rf_rates, aligned_dates

calculate_sharpe_ratio_advanced(returns, daily_risk_free_rates, trading_days_per_year=252)

Calculate Sharpe ratio using variable daily risk-free rates.

Uses the standard (non-geometric) approach: excess_return = portfolio_return - risk_free_rate

Parameters:

Name	Type	Description	Default
returns	list[float]	List of daily portfolio returns (aligned with rf rates).	required
daily_risk_free_rates	list[float]	List of daily risk-free rates (same length as returns).	required
trading_days_per_year	int	Trading days per year for annualization (default 252).	252

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_sharpe, annual_sharpe).

Raises:

Type	Description
ValueError	If inputs have different lengths, fewer than 2 observations, or zero standard deviation.

Source code in src/athena/metrics/sharpe_advanced.py

def calculate_sharpe_ratio_advanced(
    returns: list[float],
    daily_risk_free_rates: list[float],
    trading_days_per_year: int = 252
) -> Tuple[float, float]:
    """
    Calculate Sharpe ratio using variable daily risk-free rates.

    Uses the standard (non-geometric) approach:
    excess_return = portfolio_return - risk_free_rate

    Args:
        returns: List of daily portfolio returns (aligned with rf rates).
        daily_risk_free_rates: List of daily risk-free rates (same length as returns).
        trading_days_per_year: Trading days per year for annualization (default 252).

    Returns:
        Tuple of (daily_sharpe, annual_sharpe).

    Raises:
        ValueError: If inputs have different lengths, fewer than 2 observations,
                    or zero standard deviation.
    """
    if len(returns) != len(daily_risk_free_rates):
        raise ValueError(
            f"Returns and risk-free rates must have same length. "
            f"Got {len(returns)} returns and {len(daily_risk_free_rates)} rates."
        )

    if len(returns) < 2:
        raise ValueError("Need at least two return observations.")

    # Calculate excess returns (standard approach: simple subtraction)
    portfolio_returns = np.array(returns, dtype=float)
    rf_rates = np.array(daily_risk_free_rates, dtype=float)
    excess_returns = portfolio_returns - rf_rates

    mean_excess = excess_returns.mean()
    std_excess = excess_returns.std(ddof=1)

    if np.isclose(std_excess, 0.0):
        raise ValueError("Sample standard deviation is zero; Sharpe is undefined.")

    daily_sharpe = float(mean_excess / std_excess)
    annual_sharpe = daily_sharpe * np.sqrt(trading_days_per_year)

    return daily_sharpe, annual_sharpe

calculate_sharpe_ratio_by_day_cumulative_advanced(portfolio, target_currency, rf_manager=None, start_date=None, end_date=None, trading_days_per_year=252, min_observations=2)

Calculate cumulative Sharpe ratios for each trading day in a date range.

For each trading day, calculates the Sharpe ratio using all returns from start_date up to that day (cumulative/expanding window).

Only includes dates where risk-free rate data is available.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
rf_manager	RiskFreeRateManager | None	Risk-free rate manager. If None, uses FRED DTB3.	None
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
trading_days_per_year	int	Trading days per year for annualization (default 252).	252
min_observations	int	Minimum number of observations required before calculating Sharpe. Defaults to 2.	2

Returns:

Type	Description
dict[datetime, Tuple[float, float]]	Dictionary mapping each trading day to (daily_sharpe, annual_sharpe).
dict[datetime, Tuple[float, float]]	Days with insufficient observations are excluded.

Source code in src/athena/metrics/sharpe_advanced.py

def calculate_sharpe_ratio_by_day_cumulative_advanced(
    portfolio: Portfolio,
    target_currency: Currency,
    rf_manager: RiskFreeRateManager | None = None,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    trading_days_per_year: int = 252,
    min_observations: int = 2
) -> dict[datetime, Tuple[float, float]]:
    """
    Calculate cumulative Sharpe ratios for each trading day in a date range.

    For each trading day, calculates the Sharpe ratio using all returns from
    start_date up to that day (cumulative/expanding window).

    Only includes dates where risk-free rate data is available.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        rf_manager: Risk-free rate manager. If None, uses FRED DTB3.
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        trading_days_per_year: Trading days per year for annualization (default 252).
        min_observations: Minimum number of observations required before
                          calculating Sharpe. Defaults to 2.

    Returns:
        Dictionary mapping each trading day to (daily_sharpe, annual_sharpe).
        Days with insufficient observations are excluded.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < min_observations:
        return {}

    # Determine date range for risk-free rate data
    return_dates = sorted(daily_returns.keys())
    min_dt = return_dates[0].date()
    max_dt = return_dates[-1].date()

    # Create default manager if not provided
    if rf_manager is None:
        rf_manager = _get_default_risk_free_manager(min_dt, max_dt)

    # Align returns with risk-free rates (trading days only)
    aligned_returns, aligned_rf_rates, aligned_dates = align_returns_with_risk_free_rates(
        daily_returns,
        rf_manager,
        trading_days_per_year
    )

    if len(aligned_returns) < min_observations:
        return {}

    # Calculate cumulative Sharpe for each trading day
    result: dict[datetime, Tuple[float, float]] = {}
    cumulative_returns: list[float] = []
    cumulative_rf_rates: list[float] = []

    for i, dt in enumerate(aligned_dates):
        cumulative_returns.append(aligned_returns[i])
        cumulative_rf_rates.append(aligned_rf_rates[i])

        if len(cumulative_returns) < min_observations:
            continue

        excess_returns = np.array(cumulative_returns) - np.array(cumulative_rf_rates)
        mean_excess = excess_returns.mean()
        std_excess = excess_returns.std(ddof=1)

        if np.isclose(std_excess, 0.0):
            continue

        daily_sharpe = float(mean_excess / std_excess)
        annual_sharpe = daily_sharpe * np.sqrt(trading_days_per_year)

        result[dt] = (daily_sharpe, annual_sharpe)

    return result

calculate_sharpe_ratio_by_day_rolling_window_advanced(portfolio, target_currency, window_size, rf_manager=None, start_date=None, end_date=None, trading_days_per_year=252)

Calculate rolling window Sharpe ratios for each trading day.

For each trading day, calculates the Sharpe ratio using the trailing window_size trading days of returns.

Only includes dates where risk-free rate data is available.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
window_size	int	Number of trailing trading days to use for each calculation. Common values: 21 (month), 63 (quarter), 252 (year).	required
rf_manager	RiskFreeRateManager | None	Risk-free rate manager. If None, uses FRED DTB3.	None
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
trading_days_per_year	int	Trading days per year for annualization (default 252).	252

Returns:

Type	Description
dict[datetime, Tuple[float, float]]	Dictionary mapping each trading day to (daily_sharpe, annual_sharpe).
dict[datetime, Tuple[float, float]]	Days with insufficient observations (fewer than window_size) are excluded.

Source code in src/athena/metrics/sharpe_advanced.py

def calculate_sharpe_ratio_by_day_rolling_window_advanced(
    portfolio: Portfolio,
    target_currency: Currency,
    window_size: int,
    rf_manager: RiskFreeRateManager | None = None,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    trading_days_per_year: int = 252
) -> dict[datetime, Tuple[float, float]]:
    """
    Calculate rolling window Sharpe ratios for each trading day.

    For each trading day, calculates the Sharpe ratio using the trailing
    window_size trading days of returns.

    Only includes dates where risk-free rate data is available.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        window_size: Number of trailing trading days to use for each calculation.
                     Common values: 21 (month), 63 (quarter), 252 (year).
        rf_manager: Risk-free rate manager. If None, uses FRED DTB3.
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        trading_days_per_year: Trading days per year for annualization (default 252).

    Returns:
        Dictionary mapping each trading day to (daily_sharpe, annual_sharpe).
        Days with insufficient observations (fewer than window_size) are excluded.
    """
    if window_size < 2:
        raise ValueError("window_size must be at least 2.")

    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < window_size:
        return {}

    # Determine date range for risk-free rate data
    return_dates = sorted(daily_returns.keys())
    min_dt = return_dates[0].date()
    max_dt = return_dates[-1].date()

    # Create default manager if not provided
    if rf_manager is None:
        rf_manager = _get_default_risk_free_manager(min_dt, max_dt)

    # Align returns with risk-free rates (trading days only)
    aligned_returns, aligned_rf_rates, aligned_dates = align_returns_with_risk_free_rates(
        daily_returns,
        rf_manager,
        trading_days_per_year
    )

    if len(aligned_returns) < window_size:
        return {}

    # Calculate rolling Sharpe for each trading day
    result: dict[datetime, Tuple[float, float]] = {}

    for i in range(window_size - 1, len(aligned_dates)):
        window_returns = aligned_returns[i - window_size + 1:i + 1]
        window_rf_rates = aligned_rf_rates[i - window_size + 1:i + 1]
        dt = aligned_dates[i]

        excess_returns = np.array(window_returns) - np.array(window_rf_rates)
        mean_excess = excess_returns.mean()
        std_excess = excess_returns.std(ddof=1)

        if np.isclose(std_excess, 0.0):
            continue

        daily_sharpe = float(mean_excess / std_excess)
        annual_sharpe = daily_sharpe * np.sqrt(trading_days_per_year)

        result[dt] = (daily_sharpe, annual_sharpe)

    return result

calculate_sharpe_ratio_cumulative_advanced(portfolio, target_currency, rf_manager=None, start_date=None, end_date=None, trading_days_per_year=252)

Calculate the cumulative Sharpe ratio using dynamic risk-free rates.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
rf_manager	RiskFreeRateManager | None	Risk-free rate manager. If None, uses FRED DTB3.	None
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
trading_days_per_year	int	Trading days per year for annualization (default 252).	252

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_sharpe, annual_sharpe).

Raises:

Type	Description
ValueError	If insufficient data or calculation error.

Source code in src/athena/metrics/sharpe_advanced.py

def calculate_sharpe_ratio_cumulative_advanced(
    portfolio: Portfolio,
    target_currency: Currency,
    rf_manager: RiskFreeRateManager | None = None,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    trading_days_per_year: int = 252
) -> Tuple[float, float]:
    """
    Calculate the cumulative Sharpe ratio using dynamic risk-free rates.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        rf_manager: Risk-free rate manager. If None, uses FRED DTB3.
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        trading_days_per_year: Trading days per year for annualization (default 252).

    Returns:
        Tuple of (daily_sharpe, annual_sharpe).

    Raises:
        ValueError: If insufficient data or calculation error.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        raise ValueError("Need at least two portfolio value observations.")

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < 2:
        raise ValueError("Need at least two return observations.")

    # Determine date range for risk-free rate data
    return_dates = sorted(daily_returns.keys())
    min_date = return_dates[0].date()
    max_date = return_dates[-1].date()

    # Create default manager if not provided
    if rf_manager is None:
        rf_manager = _get_default_risk_free_manager(min_date, max_date)

    # Align returns with risk-free rates (trading days only)
    aligned_returns, aligned_rf_rates, _ = align_returns_with_risk_free_rates(
        daily_returns,
        rf_manager,
        trading_days_per_year
    )

    if len(aligned_returns) < 2:
        raise ValueError(
            "Need at least two aligned observations after filtering to trading days."
        )

    return calculate_sharpe_ratio_advanced(
        aligned_returns,
        aligned_rf_rates,
        trading_days_per_year
    )

calculate_sharpe_ratio_from_values_advanced(portfolio_values, rf_manager=None, trading_days_per_year=252)

Calculate Sharpe ratios directly from portfolio values using dynamic RF rates.

Convenience function when you already have portfolio values and don't need to calculate them from a Portfolio object.

Parameters:

Name	Type	Description	Default
portfolio_values	dict[datetime, Decimal]	Dictionary mapping dates to portfolio values.	required
rf_manager	RiskFreeRateManager | None	Risk-free rate manager. If None, uses FRED DTB3.	None
trading_days_per_year	int	Trading days per year for annualization (default 252).	252

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_sharpe, annual_sharpe).

Raises:

Type	Description
ValueError	If insufficient data or calculation error.

Source code in src/athena/metrics/sharpe_advanced.py

def calculate_sharpe_ratio_from_values_advanced(
    portfolio_values: dict[datetime, Decimal],
    rf_manager: RiskFreeRateManager | None = None,
    trading_days_per_year: int = 252
) -> Tuple[float, float]:
    """
    Calculate Sharpe ratios directly from portfolio values using dynamic RF rates.

    Convenience function when you already have portfolio values and don't
    need to calculate them from a Portfolio object.

    Args:
        portfolio_values: Dictionary mapping dates to portfolio values.
        rf_manager: Risk-free rate manager. If None, uses FRED DTB3.
        trading_days_per_year: Trading days per year for annualization (default 252).

    Returns:
        Tuple of (daily_sharpe, annual_sharpe).

    Raises:
        ValueError: If insufficient data or calculation error.
    """
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < 2:
        raise ValueError("Need at least two return observations.")

    # Determine date range for risk-free rate data
    return_dates = sorted(daily_returns.keys())
    min_date = return_dates[0].date()
    max_date = return_dates[-1].date()

    # Create default manager if not provided
    if rf_manager is None:
        rf_manager = _get_default_risk_free_manager(min_date, max_date)

    # Align returns with risk-free rates
    aligned_returns, aligned_rf_rates, _ = align_returns_with_risk_free_rates(
        daily_returns,
        rf_manager,
        trading_days_per_year
    )

    if len(aligned_returns) < 2:
        raise ValueError(
            "Need at least two aligned observations after filtering to trading days."
        )

    return calculate_sharpe_ratio_advanced(
        aligned_returns,
        aligned_rf_rates,
        trading_days_per_year
    )

calculate_downside_deviation(returns, target_return=0.0)

Calculate downside deviation from a list of returns.

Downside deviation measures the volatility of returns that fall below a target return (typically the risk-free rate or zero).

Parameters:

Name	Type	Description	Default
returns	list[float]	List of returns as decimals (e.g., 0.002 = 0.2%).	required
target_return	float	The minimum acceptable return (default 0.0).	0.0

Returns:

Type	Description
float	The downside deviation as a float.

Raises:

Type	Description
ValueError	If fewer than two observations.

Source code in src/athena/metrics/sortino.py

def calculate_downside_deviation(
    returns: list[float],
    target_return: float = 0.0
) -> float:
    """
    Calculate downside deviation from a list of returns.

    Downside deviation measures the volatility of returns that fall below
    a target return (typically the risk-free rate or zero).

    Args:
        returns: List of returns as decimals (e.g., 0.002 = 0.2%).
        target_return: The minimum acceptable return (default 0.0).

    Returns:
        The downside deviation as a float.

    Raises:
        ValueError: If fewer than two observations.
    """
    if len(returns) < 2:
        raise ValueError("Need at least two return observations.")

    returns_array = np.array(returns, dtype=float)
    downside_returns = np.minimum(returns_array - target_return, 0.0)
    downside_deviation = np.sqrt(np.mean(downside_returns ** 2))

    return float(downside_deviation)

calculate_sortino_ratio(returns, annual_risk_free_rate, periods_in_year=365)

Calculate daily and annualized Sortino ratios from a list of returns.

The Sortino ratio is similar to the Sharpe ratio but only considers downside volatility, making it a better measure for investors who are primarily concerned with downside risk.

Parameters:

Name	Type	Description	Default
returns	list[float]	List of daily returns as decimals (e.g., 0.002 = 0.2%).	required
annual_risk_free_rate	float	Annual nominal risk-free rate (e.g., 0.03 for 3%).	required
periods_in_year	int	Trading periods in a year (365 for daily).	365

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_sortino, annual_sortino).

Raises:

Type	Description
ValueError	If fewer than two observations or zero downside deviation.

Source code in src/athena/metrics/sortino.py

def calculate_sortino_ratio(
    returns: list[float],
    annual_risk_free_rate: float,
    periods_in_year: int = 365
) -> Tuple[float, float]:
    """
    Calculate daily and annualized Sortino ratios from a list of returns.

    The Sortino ratio is similar to the Sharpe ratio but only considers
    downside volatility, making it a better measure for investors who
    are primarily concerned with downside risk.

    Args:
        returns: List of daily returns as decimals (e.g., 0.002 = 0.2%).
        annual_risk_free_rate: Annual nominal risk-free rate (e.g., 0.03 for 3%).
        periods_in_year: Trading periods in a year (365 for daily).

    Returns:
        Tuple of (daily_sortino, annual_sortino).

    Raises:
        ValueError: If fewer than two observations or zero downside deviation.
    """
    if len(returns) < 2:
        raise ValueError("Need at least two return observations.")

    daily_rf = annual_risk_free_rate / periods_in_year

    returns_array = np.array(returns, dtype=float)
    mean_return = returns_array.mean()
    excess_return = mean_return - daily_rf

    downside_deviation = calculate_downside_deviation(returns, daily_rf)

    if np.isclose(downside_deviation, 0.0):
        raise ValueError("Downside deviation is zero; Sortino is undefined.")

    daily_sortino = float(excess_return / downside_deviation)
    annual_sortino = daily_sortino * np.sqrt(periods_in_year)

    return daily_sortino, annual_sortino

calculate_sortino_ratio_by_day_cumulative(portfolio, target_currency, annual_risk_free_rate, start_date=None, end_date=None, periods_in_year=365, min_observations=2)

Calculate cumulative Sortino ratios for each day in a date range.

For each day, calculates the Sortino ratio using all returns from start_date up to that day (cumulative/expanding window).

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
annual_risk_free_rate	float	Annual nominal risk-free rate (e.g., 0.03 for 3%).	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
periods_in_year	int	Trading periods in a year (365 for daily, 252 for US equities trading days).	365
min_observations	int	Minimum number of return observations required before calculating Sortino. Defaults to 2.	2

Returns:

Type	Description
dict[datetime, Tuple[float, float]]	Dictionary mapping each date to a tuple of (daily_sortino, annual_sortino).
dict[datetime, Tuple[float, float]]	Days with insufficient observations are excluded.

Source code in src/athena/metrics/sortino.py

def calculate_sortino_ratio_by_day_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    annual_risk_free_rate: float,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    periods_in_year: int = 365,
    min_observations: int = 2
) -> dict[datetime, Tuple[float, float]]:
    """
    Calculate cumulative Sortino ratios for each day in a date range.

    For each day, calculates the Sortino ratio using all returns from
    start_date up to that day (cumulative/expanding window).

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        annual_risk_free_rate: Annual nominal risk-free rate (e.g., 0.03 for 3%).
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        periods_in_year: Trading periods in a year (365 for daily, 252 for
                         US equities trading days).
        min_observations: Minimum number of return observations required
                          before calculating Sortino. Defaults to 2.

    Returns:
        Dictionary mapping each date to a tuple of (daily_sortino, annual_sortino).
        Days with insufficient observations are excluded.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < min_observations:
        return {}

    # Calculate cumulative Sortino for each day
    sorted_dates = sorted(daily_returns.keys())
    daily_rf = annual_risk_free_rate / periods_in_year

    result: dict[datetime, Tuple[float, float]] = {}
    cumulative_returns: list[float] = []

    for date in sorted_dates:
        cumulative_returns.append(daily_returns[date])

        if len(cumulative_returns) < min_observations:
            continue

        returns_array = np.array(cumulative_returns, dtype=float)
        mean_return = returns_array.mean()
        excess_return = mean_return - daily_rf

        downside_returns = np.minimum(returns_array - daily_rf, 0.0)
        downside_deviation = np.sqrt(np.mean(downside_returns ** 2))

        if np.isclose(downside_deviation, 0.0):
            # Skip days where downside deviation is zero
            continue

        daily_sortino = float(excess_return / downside_deviation)
        annual_sortino = daily_sortino * np.sqrt(periods_in_year)

        result[date] = (daily_sortino, annual_sortino)

    return result

calculate_sortino_ratio_by_day_rolling_window(portfolio, target_currency, annual_risk_free_rate, window_size, start_date=None, end_date=None, periods_in_year=365)

Calculate rolling window Sortino ratios for each day in a date range.

For each day, calculates the Sortino ratio using the trailing window_size days of returns.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
annual_risk_free_rate	float	Annual nominal risk-free rate (e.g., 0.03 for 3%).	required
window_size	int	Number of trailing days to use for each calculation. Common values: 30, 60, 90, 252 (trading year).	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
periods_in_year	int	Trading periods in a year (365 for daily, 252 for US equities trading days).	365

Returns:

Type	Description
dict[datetime, Tuple[float, float]]	Dictionary mapping each date to a tuple of (daily_sortino, annual_sortino).
dict[datetime, Tuple[float, float]]	Days with insufficient observations (fewer than window_size) are excluded.

Source code in src/athena/metrics/sortino.py

def calculate_sortino_ratio_by_day_rolling_window(
    portfolio: Portfolio,
    target_currency: Currency,
    annual_risk_free_rate: float,
    window_size: int,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    periods_in_year: int = 365
) -> dict[datetime, Tuple[float, float]]:
    """
    Calculate rolling window Sortino ratios for each day in a date range.

    For each day, calculates the Sortino ratio using the trailing
    window_size days of returns.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        annual_risk_free_rate: Annual nominal risk-free rate (e.g., 0.03 for 3%).
        window_size: Number of trailing days to use for each calculation.
                     Common values: 30, 60, 90, 252 (trading year).
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        periods_in_year: Trading periods in a year (365 for daily, 252 for
                         US equities trading days).

    Returns:
        Dictionary mapping each date to a tuple of (daily_sortino, annual_sortino).
        Days with insufficient observations (fewer than window_size) are excluded.
    """
    if window_size < 2:
        raise ValueError("window_size must be at least 2.")

    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < window_size:
        return {}

    # Calculate rolling Sortino for each day
    sorted_dates = sorted(daily_returns.keys())
    returns_list = [daily_returns[d] for d in sorted_dates]
    daily_rf = annual_risk_free_rate / periods_in_year

    result: dict[datetime, Tuple[float, float]] = {}

    for i in range(window_size - 1, len(sorted_dates)):
        window_returns = returns_list[i - window_size + 1:i + 1]
        date = sorted_dates[i]

        returns_array = np.array(window_returns, dtype=float)
        mean_return = returns_array.mean()
        excess_return = mean_return - daily_rf

        downside_returns = np.minimum(returns_array - daily_rf, 0.0)
        downside_deviation = np.sqrt(np.mean(downside_returns ** 2))

        if np.isclose(downside_deviation, 0.0):
            # Skip days where downside deviation is zero
            continue

        daily_sortino = float(excess_return / downside_deviation)
        annual_sortino = daily_sortino * np.sqrt(periods_in_year)

        result[date] = (daily_sortino, annual_sortino)

    return result

calculate_sortino_ratio_cumulative(portfolio, target_currency, annual_risk_free_rate, start_date=None, end_date=None, periods_in_year=365)

Calculate the cumulative Sortino ratio for a portfolio over a date range.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
annual_risk_free_rate	float	Annual nominal risk-free rate (e.g., 0.03 for 3%).	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
periods_in_year	int	Trading periods in a year (365 for daily, 252 for US equities trading days).	365

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_sortino, annual_sortino).

Raises:

Type	Description
ValueError	If fewer than two observations or zero downside deviation.

Source code in src/athena/metrics/sortino.py

def calculate_sortino_ratio_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    annual_risk_free_rate: float,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    periods_in_year: int = 365
) -> Tuple[float, float]:
    """
    Calculate the cumulative Sortino ratio for a portfolio over a date range.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        annual_risk_free_rate: Annual nominal risk-free rate (e.g., 0.03 for 3%).
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        periods_in_year: Trading periods in a year (365 for daily, 252 for
                         US equities trading days).

    Returns:
        Tuple of (daily_sortino, annual_sortino).

    Raises:
        ValueError: If fewer than two observations or zero downside deviation.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        raise ValueError("Need at least two portfolio value observations.")

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < 2:
        raise ValueError("Need at least two return observations.")

    # Calculate Sortino ratio
    returns_list = [daily_returns[d] for d in sorted(daily_returns.keys())]
    return calculate_sortino_ratio(returns_list, annual_risk_free_rate, periods_in_year)

calculate_sortino_ratio_from_values(portfolio_values, annual_risk_free_rate, periods_in_year=365)

Calculate Sortino ratios directly from a dictionary of portfolio values.

This is a convenience function when you already have portfolio values and don't need to calculate them from a Portfolio object.

Parameters:

Name	Type	Description	Default
portfolio_values	dict[datetime, Decimal]	Dictionary mapping dates to portfolio values.	required
annual_risk_free_rate	float	Annual nominal risk-free rate (e.g., 0.03 for 3%).	required
periods_in_year	int	Trading periods in a year (365 for daily).	365

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_sortino, annual_sortino).

Raises:

Type	Description
ValueError	If fewer than two observations or zero downside deviation.

Source code in src/athena/metrics/sortino.py

def calculate_sortino_ratio_from_values(
    portfolio_values: dict[datetime, Decimal],
    annual_risk_free_rate: float,
    periods_in_year: int = 365
) -> Tuple[float, float]:
    """
    Calculate Sortino ratios directly from a dictionary of portfolio values.

    This is a convenience function when you already have portfolio values
    and don't need to calculate them from a Portfolio object.

    Args:
        portfolio_values: Dictionary mapping dates to portfolio values.
        annual_risk_free_rate: Annual nominal risk-free rate (e.g., 0.03 for 3%).
        periods_in_year: Trading periods in a year (365 for daily).

    Returns:
        Tuple of (daily_sortino, annual_sortino).

    Raises:
        ValueError: If fewer than two observations or zero downside deviation.
    """
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < 2:
        raise ValueError("Need at least two return observations.")

    returns_list = [daily_returns[d] for d in sorted(daily_returns.keys())]

    return calculate_sortino_ratio(returns_list, annual_risk_free_rate, periods_in_year)

calculate_cvar(returns, confidence_level=0.95)

Calculate Conditional Value at Risk (Expected Shortfall).

CVaR represents the expected loss given that the loss exceeds the VaR threshold. It provides a more complete picture of tail risk than VaR alone.

Parameters:

Name	Type	Description	Default
returns	list[float]	List of daily returns as decimals (e.g., 0.002 = 0.2%).	required
confidence_level	float	Confidence level for CVaR (e.g., 0.95 for 95%).	0.95

Returns:

Type	Description
float	CVaR as a negative float representing the expected loss beyond VaR
float	(e.g., -0.08 means 8% expected loss in worst cases).

Raises:

Type	Description
ValueError	If fewer than two observations or invalid confidence level.

Source code in src/athena/metrics/value_at_risk.py

def calculate_cvar(
    returns: list[float],
    confidence_level: float = 0.95
) -> float:
    """
    Calculate Conditional Value at Risk (Expected Shortfall).

    CVaR represents the expected loss given that the loss exceeds the VaR
    threshold. It provides a more complete picture of tail risk than VaR alone.

    Args:
        returns: List of daily returns as decimals (e.g., 0.002 = 0.2%).
        confidence_level: Confidence level for CVaR (e.g., 0.95 for 95%).

    Returns:
        CVaR as a negative float representing the expected loss beyond VaR
        (e.g., -0.08 means 8% expected loss in worst cases).

    Raises:
        ValueError: If fewer than two observations or invalid confidence level.
    """
    if len(returns) < 2:
        raise ValueError("Need at least two return observations.")

    if not 0 < confidence_level < 1:
        raise ValueError("Confidence level must be between 0 and 1.")

    returns_array = np.array(returns, dtype=float)
    var = calculate_var_historical(returns, confidence_level)

    # CVaR is the mean of returns that are worse than VaR
    tail_returns = returns_array[returns_array <= var]

    if len(tail_returns) == 0:
        return var

    return float(tail_returns.mean())

calculate_var_by_day_cumulative(portfolio, target_currency, confidence_level=0.95, method='historical', start_date=None, end_date=None, min_observations=30)

Calculate cumulative Value at Risk for each day in a date range.

For each day, calculates VaR using all returns from start_date up to that day (cumulative/expanding window).

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
confidence_level	float	Confidence level for VaR (e.g., 0.95 for 95%).	0.95
method	str	VaR calculation method ("historical" or "parametric").	'historical'
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
min_observations	int	Minimum number of return observations required before calculating VaR. Defaults to 30.	30

Returns:

Type	Description
dict[datetime, float]	Dictionary mapping each date to the cumulative VaR up to that date.
dict[datetime, float]	Days with insufficient observations are excluded.

Source code in src/athena/metrics/value_at_risk.py

def calculate_var_by_day_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    confidence_level: float = 0.95,
    method: str = "historical",
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    min_observations: int = 30
) -> dict[datetime, float]:
    """
    Calculate cumulative Value at Risk for each day in a date range.

    For each day, calculates VaR using all returns from start_date up to
    that day (cumulative/expanding window).

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        confidence_level: Confidence level for VaR (e.g., 0.95 for 95%).
        method: VaR calculation method ("historical" or "parametric").
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        min_observations: Minimum number of return observations required
                          before calculating VaR. Defaults to 30.

    Returns:
        Dictionary mapping each date to the cumulative VaR up to that date.
        Days with insufficient observations are excluded.
    """
    if method not in ("historical", "parametric"):
        raise ValueError("Method must be 'historical' or 'parametric'.")

    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < min_observations:
        return {}

    sorted_dates = sorted(daily_returns.keys())
    result: dict[datetime, float] = {}
    cumulative_returns: list[float] = []

    var_func = calculate_var_historical if method == "historical" else calculate_var_parametric

    for date in sorted_dates:
        cumulative_returns.append(daily_returns[date])

        if len(cumulative_returns) < min_observations:
            continue

        try:
            var = var_func(cumulative_returns, confidence_level)
            result[date] = var
        except ValueError:
            continue

    return result

calculate_var_by_day_rolling_window(portfolio, target_currency, window_size, confidence_level=0.95, method='historical', start_date=None, end_date=None)

Calculate rolling window Value at Risk for each day in a date range.

For each day, calculates VaR using the trailing window_size days of returns.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
window_size	int	Number of trailing days to use for each calculation. Common values: 30, 60, 90, 252 (trading year).	required
confidence_level	float	Confidence level for VaR (e.g., 0.95 for 95%).	0.95
method	str	VaR calculation method ("historical" or "parametric").	'historical'
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None

Returns:

Type	Description
dict[datetime, float]	Dictionary mapping each date to the rolling VaR.
dict[datetime, float]	Days with insufficient observations (fewer than window_size) are excluded.

Source code in src/athena/metrics/value_at_risk.py

def calculate_var_by_day_rolling_window(
    portfolio: Portfolio,
    target_currency: Currency,
    window_size: int,
    confidence_level: float = 0.95,
    method: str = "historical",
    start_date: datetime | None = None,
    end_date: datetime | None = None
) -> dict[datetime, float]:
    """
    Calculate rolling window Value at Risk for each day in a date range.

    For each day, calculates VaR using the trailing window_size days of returns.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        window_size: Number of trailing days to use for each calculation.
                     Common values: 30, 60, 90, 252 (trading year).
        confidence_level: Confidence level for VaR (e.g., 0.95 for 95%).
        method: VaR calculation method ("historical" or "parametric").
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.

    Returns:
        Dictionary mapping each date to the rolling VaR.
        Days with insufficient observations (fewer than window_size) are excluded.
    """
    if window_size < 2:
        raise ValueError("window_size must be at least 2.")

    if method not in ("historical", "parametric"):
        raise ValueError("Method must be 'historical' or 'parametric'.")

    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < window_size:
        return {}

    sorted_dates = sorted(daily_returns.keys())
    returns_list = [daily_returns[d] for d in sorted_dates]

    var_func = calculate_var_historical if method == "historical" else calculate_var_parametric

    result: dict[datetime, float] = {}

    for i in range(window_size - 1, len(sorted_dates)):
        window_returns = returns_list[i - window_size + 1:i + 1]
        date = sorted_dates[i]

        try:
            var = var_func(window_returns, confidence_level)
            result[date] = var
        except ValueError:
            continue

    return result

calculate_var_cumulative(portfolio, target_currency, confidence_level=0.95, method='historical', start_date=None, end_date=None)

Calculate Value at Risk and CVaR for a portfolio over a date range.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
confidence_level	float	Confidence level for VaR (e.g., 0.95 for 95%).	0.95
method	str	VaR calculation method ("historical" or "parametric").	'historical'
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None

Returns:

Type	Description
Tuple[float, float]	Tuple of (var, cvar) as negative floats.

Raises:

Type	Description
ValueError	If fewer than two observations or invalid parameters.

Source code in src/athena/metrics/value_at_risk.py

def calculate_var_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    confidence_level: float = 0.95,
    method: str = "historical",
    start_date: datetime | None = None,
    end_date: datetime | None = None
) -> Tuple[float, float]:
    """
    Calculate Value at Risk and CVaR for a portfolio over a date range.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        confidence_level: Confidence level for VaR (e.g., 0.95 for 95%).
        method: VaR calculation method ("historical" or "parametric").
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.

    Returns:
        Tuple of (var, cvar) as negative floats.

    Raises:
        ValueError: If fewer than two observations or invalid parameters.
    """
    if method not in ("historical", "parametric"):
        raise ValueError("Method must be 'historical' or 'parametric'.")

    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        raise ValueError("Need at least two portfolio value observations.")

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < 2:
        raise ValueError("Need at least two return observations.")

    returns_list = [daily_returns[d] for d in sorted(daily_returns.keys())]

    var_func = calculate_var_historical if method == "historical" else calculate_var_parametric
    var = var_func(returns_list, confidence_level)
    cvar = calculate_cvar(returns_list, confidence_level)

    return var, cvar

calculate_var_from_values(portfolio_values, confidence_level=0.95, method='historical')

Calculate VaR and CVaR directly from a dictionary of portfolio values.

This is a convenience function when you already have portfolio values and don't need to calculate them from a Portfolio object.

Parameters:

Name	Type	Description	Default
portfolio_values	dict[datetime, Decimal]	Dictionary mapping dates to portfolio values.	required
confidence_level	float	Confidence level for VaR (e.g., 0.95 for 95%).	0.95
method	str	VaR calculation method ("historical" or "parametric").	'historical'

Returns:

Type	Description
Tuple[float, float]	Tuple of (var, cvar) as negative floats.

Raises:

Type	Description
ValueError	If fewer than two observations or invalid parameters.

Source code in src/athena/metrics/value_at_risk.py

def calculate_var_from_values(
    portfolio_values: dict[datetime, Decimal],
    confidence_level: float = 0.95,
    method: str = "historical"
) -> Tuple[float, float]:
    """
    Calculate VaR and CVaR directly from a dictionary of portfolio values.

    This is a convenience function when you already have portfolio values
    and don't need to calculate them from a Portfolio object.

    Args:
        portfolio_values: Dictionary mapping dates to portfolio values.
        confidence_level: Confidence level for VaR (e.g., 0.95 for 95%).
        method: VaR calculation method ("historical" or "parametric").

    Returns:
        Tuple of (var, cvar) as negative floats.

    Raises:
        ValueError: If fewer than two observations or invalid parameters.
    """
    if method not in ("historical", "parametric"):
        raise ValueError("Method must be 'historical' or 'parametric'.")

    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < 2:
        raise ValueError("Need at least two return observations.")

    returns_list = [daily_returns[d] for d in sorted(daily_returns.keys())]

    var_func = calculate_var_historical if method == "historical" else calculate_var_parametric
    var = var_func(returns_list, confidence_level)
    cvar = calculate_cvar(returns_list, confidence_level)

    return var, cvar

calculate_var_historical(returns, confidence_level=0.95)

Calculate Value at Risk using the historical simulation method.

Historical VaR uses the actual distribution of past returns to estimate potential losses at a given confidence level.

Parameters:

Name	Type	Description	Default
returns	list[float]	List of daily returns as decimals (e.g., 0.002 = 0.2%).	required
confidence_level	float	Confidence level for VaR (e.g., 0.95 for 95%).	0.95

Returns:

Type	Description
float	VaR as a negative float representing the potential loss
float	(e.g., -0.05 means 5% potential loss at the given confidence level).

Raises:

Type	Description
ValueError	If fewer than two observations or invalid confidence level.

Source code in src/athena/metrics/value_at_risk.py

def calculate_var_historical(
    returns: list[float],
    confidence_level: float = 0.95
) -> float:
    """
    Calculate Value at Risk using the historical simulation method.

    Historical VaR uses the actual distribution of past returns to estimate
    potential losses at a given confidence level.

    Args:
        returns: List of daily returns as decimals (e.g., 0.002 = 0.2%).
        confidence_level: Confidence level for VaR (e.g., 0.95 for 95%).

    Returns:
        VaR as a negative float representing the potential loss
        (e.g., -0.05 means 5% potential loss at the given confidence level).

    Raises:
        ValueError: If fewer than two observations or invalid confidence level.
    """
    if len(returns) < 2:
        raise ValueError("Need at least two return observations.")

    if not 0 < confidence_level < 1:
        raise ValueError("Confidence level must be between 0 and 1.")

    returns_array = np.array(returns, dtype=float)
    var = float(np.percentile(returns_array, (1 - confidence_level) * 100))

    return var

calculate_var_parametric(returns, confidence_level=0.95)

Calculate Value at Risk using the parametric (variance-covariance) method.

Parametric VaR assumes returns are normally distributed and uses the mean and standard deviation to estimate potential losses.

Parameters:

Name	Type	Description	Default
returns	list[float]	List of daily returns as decimals (e.g., 0.002 = 0.2%).	required
confidence_level	float	Confidence level for VaR (e.g., 0.95 for 95%).	0.95

Returns:

Type	Description
float	VaR as a negative float representing the potential loss
float	(e.g., -0.05 means 5% potential loss at the given confidence level).

Raises:

Type	Description
ValueError	If fewer than two observations or invalid confidence level.

Source code in src/athena/metrics/value_at_risk.py

def calculate_var_parametric(
    returns: list[float],
    confidence_level: float = 0.95
) -> float:
    """
    Calculate Value at Risk using the parametric (variance-covariance) method.

    Parametric VaR assumes returns are normally distributed and uses the
    mean and standard deviation to estimate potential losses.

    Args:
        returns: List of daily returns as decimals (e.g., 0.002 = 0.2%).
        confidence_level: Confidence level for VaR (e.g., 0.95 for 95%).

    Returns:
        VaR as a negative float representing the potential loss
        (e.g., -0.05 means 5% potential loss at the given confidence level).

    Raises:
        ValueError: If fewer than two observations or invalid confidence level.
    """
    if len(returns) < 2:
        raise ValueError("Need at least two return observations.")

    if not 0 < confidence_level < 1:
        raise ValueError("Confidence level must be between 0 and 1.")

    returns_array = np.array(returns, dtype=float)
    mean_return = returns_array.mean()
    std_return = returns_array.std(ddof=1)

    # Get the z-score for the confidence level
    z_score = stats.norm.ppf(1 - confidence_level)
    var = float(mean_return + z_score * std_return)

    return var

calculate_downside_volatility(returns, periods_in_year=365)

Calculate downside volatility (volatility of negative returns only).

Downside volatility measures the dispersion of returns below zero, capturing the variability of losses.

Parameters:

Name	Type	Description	Default
returns	list[float]	List of daily returns as decimals (e.g., 0.002 = 0.2%).	required
periods_in_year	int	Trading periods in a year (365 for daily).	365

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_downside_volatility, annual_downside_volatility).

Raises:

Type	Description
ValueError	If fewer than two negative return observations.

Source code in src/athena/metrics/volatility.py

def calculate_downside_volatility(
    returns: list[float],
    periods_in_year: int = 365
) -> Tuple[float, float]:
    """
    Calculate downside volatility (volatility of negative returns only).

    Downside volatility measures the dispersion of returns below zero,
    capturing the variability of losses.

    Args:
        returns: List of daily returns as decimals (e.g., 0.002 = 0.2%).
        periods_in_year: Trading periods in a year (365 for daily).

    Returns:
        Tuple of (daily_downside_volatility, annual_downside_volatility).

    Raises:
        ValueError: If fewer than two negative return observations.
    """
    negative_returns = [r for r in returns if r < 0]

    if len(negative_returns) < 2:
        raise ValueError("Need at least two negative return observations.")

    returns_array = np.array(negative_returns, dtype=float)
    daily_volatility = float(returns_array.std(ddof=1))
    annual_volatility = daily_volatility * np.sqrt(periods_in_year)

    return daily_volatility, annual_volatility

calculate_upside_volatility(returns, periods_in_year=365)

Calculate upside volatility (volatility of positive returns only).

Upside volatility measures the dispersion of returns above zero, capturing the variability of gains.

Parameters:

Name	Type	Description	Default
returns	list[float]	List of daily returns as decimals (e.g., 0.002 = 0.2%).	required
periods_in_year	int	Trading periods in a year (365 for daily).	365

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_upside_volatility, annual_upside_volatility).

Raises:

Type	Description
ValueError	If fewer than two positive return observations.

Source code in src/athena/metrics/volatility.py

def calculate_upside_volatility(
    returns: list[float],
    periods_in_year: int = 365
) -> Tuple[float, float]:
    """
    Calculate upside volatility (volatility of positive returns only).

    Upside volatility measures the dispersion of returns above zero,
    capturing the variability of gains.

    Args:
        returns: List of daily returns as decimals (e.g., 0.002 = 0.2%).
        periods_in_year: Trading periods in a year (365 for daily).

    Returns:
        Tuple of (daily_upside_volatility, annual_upside_volatility).

    Raises:
        ValueError: If fewer than two positive return observations.
    """
    positive_returns = [r for r in returns if r > 0]

    if len(positive_returns) < 2:
        raise ValueError("Need at least two positive return observations.")

    returns_array = np.array(positive_returns, dtype=float)
    daily_volatility = float(returns_array.std(ddof=1))
    annual_volatility = daily_volatility * np.sqrt(periods_in_year)

    return daily_volatility, annual_volatility

calculate_volatility(returns, periods_in_year=365)

Calculate daily and annualized volatility from a list of returns.

Volatility is measured as the standard deviation of returns, which quantifies the dispersion of returns around their mean.

Parameters:

Name	Type	Description	Default
returns	list[float]	List of daily returns as decimals (e.g., 0.002 = 0.2%).	required
periods_in_year	int	Trading periods in a year (365 for daily).	365

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_volatility, annual_volatility).

Raises:

Type	Description
ValueError	If fewer than two observations.

Source code in src/athena/metrics/volatility.py

def calculate_volatility(
    returns: list[float],
    periods_in_year: int = 365
) -> Tuple[float, float]:
    """
    Calculate daily and annualized volatility from a list of returns.

    Volatility is measured as the standard deviation of returns, which
    quantifies the dispersion of returns around their mean.

    Args:
        returns: List of daily returns as decimals (e.g., 0.002 = 0.2%).
        periods_in_year: Trading periods in a year (365 for daily).

    Returns:
        Tuple of (daily_volatility, annual_volatility).

    Raises:
        ValueError: If fewer than two observations.
    """
    if len(returns) < 2:
        raise ValueError("Need at least two return observations.")

    returns_array = np.array(returns, dtype=float)
    daily_volatility = float(returns_array.std(ddof=1))
    annual_volatility = daily_volatility * np.sqrt(periods_in_year)

    return daily_volatility, annual_volatility

calculate_volatility_by_day_cumulative(portfolio, target_currency, start_date=None, end_date=None, periods_in_year=365, min_observations=2)

Calculate cumulative volatility for each day in a date range.

For each day, calculates the volatility using all returns from start_date up to that day (cumulative/expanding window).

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
periods_in_year	int	Trading periods in a year (365 for daily, 252 for US equities trading days).	365
min_observations	int	Minimum number of return observations required before calculating volatility. Defaults to 2.	2

Returns:

Type	Description
dict[datetime, Tuple[float, float]]	Dictionary mapping each date to a tuple of (daily_volatility, annual_volatility).
dict[datetime, Tuple[float, float]]	Days with insufficient observations are excluded.

Source code in src/athena/metrics/volatility.py

def calculate_volatility_by_day_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    periods_in_year: int = 365,
    min_observations: int = 2
) -> dict[datetime, Tuple[float, float]]:
    """
    Calculate cumulative volatility for each day in a date range.

    For each day, calculates the volatility using all returns from
    start_date up to that day (cumulative/expanding window).

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        periods_in_year: Trading periods in a year (365 for daily, 252 for
                         US equities trading days).
        min_observations: Minimum number of return observations required
                          before calculating volatility. Defaults to 2.

    Returns:
        Dictionary mapping each date to a tuple of (daily_volatility, annual_volatility).
        Days with insufficient observations are excluded.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < min_observations:
        return {}

    sorted_dates = sorted(daily_returns.keys())
    result: dict[datetime, Tuple[float, float]] = {}
    cumulative_returns: list[float] = []

    for date in sorted_dates:
        cumulative_returns.append(daily_returns[date])

        if len(cumulative_returns) < min_observations:
            continue

        returns_array = np.array(cumulative_returns, dtype=float)
        daily_volatility = float(returns_array.std(ddof=1))
        annual_volatility = daily_volatility * np.sqrt(periods_in_year)

        result[date] = (daily_volatility, annual_volatility)

    return result

calculate_volatility_by_day_rolling_window(portfolio, target_currency, window_size, start_date=None, end_date=None, periods_in_year=365)

Calculate rolling window volatility for each day in a date range.

For each day, calculates the volatility using the trailing window_size days of returns.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
window_size	int	Number of trailing days to use for each calculation. Common values: 20, 30, 60, 90.	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
periods_in_year	int	Trading periods in a year (365 for daily, 252 for US equities trading days).	365

Returns:

Type	Description
dict[datetime, Tuple[float, float]]	Dictionary mapping each date to a tuple of (daily_volatility, annual_volatility).
dict[datetime, Tuple[float, float]]	Days with insufficient observations (fewer than window_size) are excluded.

Source code in src/athena/metrics/volatility.py

def calculate_volatility_by_day_rolling_window(
    portfolio: Portfolio,
    target_currency: Currency,
    window_size: int,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    periods_in_year: int = 365
) -> dict[datetime, Tuple[float, float]]:
    """
    Calculate rolling window volatility for each day in a date range.

    For each day, calculates the volatility using the trailing
    window_size days of returns.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        window_size: Number of trailing days to use for each calculation.
                     Common values: 20, 30, 60, 90.
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        periods_in_year: Trading periods in a year (365 for daily, 252 for
                         US equities trading days).

    Returns:
        Dictionary mapping each date to a tuple of (daily_volatility, annual_volatility).
        Days with insufficient observations (fewer than window_size) are excluded.
    """
    if window_size < 2:
        raise ValueError("window_size must be at least 2.")

    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {}

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < window_size:
        return {}

    sorted_dates = sorted(daily_returns.keys())
    returns_list = [daily_returns[d] for d in sorted_dates]

    result: dict[datetime, Tuple[float, float]] = {}

    for i in range(window_size - 1, len(sorted_dates)):
        window_returns = returns_list[i - window_size + 1:i + 1]
        date = sorted_dates[i]

        returns_array = np.array(window_returns, dtype=float)
        daily_volatility = float(returns_array.std(ddof=1))
        annual_volatility = daily_volatility * np.sqrt(periods_in_year)

        result[date] = (daily_volatility, annual_volatility)

    return result

calculate_volatility_cumulative(portfolio, target_currency, start_date=None, end_date=None, periods_in_year=365)

Calculate the volatility for a portfolio over a date range.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
periods_in_year	int	Trading periods in a year (365 for daily, 252 for US equities trading days).	365

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_volatility, annual_volatility).

Raises:

Type	Description
ValueError	If fewer than two observations.

Source code in src/athena/metrics/volatility.py

def calculate_volatility_cumulative(
    portfolio: Portfolio,
    target_currency: Currency,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    periods_in_year: int = 365
) -> Tuple[float, float]:
    """
    Calculate the volatility for a portfolio over a date range.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        periods_in_year: Trading periods in a year (365 for daily, 252 for
                         US equities trading days).

    Returns:
        Tuple of (daily_volatility, annual_volatility).

    Raises:
        ValueError: If fewer than two observations.
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        raise ValueError("Need at least two portfolio value observations.")

    # Calculate daily returns
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < 2:
        raise ValueError("Need at least two return observations.")

    returns_list = [daily_returns[d] for d in sorted(daily_returns.keys())]
    return calculate_volatility(returns_list, periods_in_year)

calculate_volatility_from_values(portfolio_values, periods_in_year=365)

Calculate volatility directly from a dictionary of portfolio values.

This is a convenience function when you already have portfolio values and don't need to calculate them from a Portfolio object.

Parameters:

Name	Type	Description	Default
portfolio_values	dict[datetime, Decimal]	Dictionary mapping dates to portfolio values.	required
periods_in_year	int	Trading periods in a year (365 for daily).	365

Returns:

Type	Description
Tuple[float, float]	Tuple of (daily_volatility, annual_volatility).

Raises:

Type	Description
ValueError	If fewer than two observations.

Source code in src/athena/metrics/volatility.py

def calculate_volatility_from_values(
    portfolio_values: dict[datetime, Decimal],
    periods_in_year: int = 365
) -> Tuple[float, float]:
    """
    Calculate volatility directly from a dictionary of portfolio values.

    This is a convenience function when you already have portfolio values
    and don't need to calculate them from a Portfolio object.

    Args:
        portfolio_values: Dictionary mapping dates to portfolio values.
        periods_in_year: Trading periods in a year (365 for daily).

    Returns:
        Tuple of (daily_volatility, annual_volatility).

    Raises:
        ValueError: If fewer than two observations.
    """
    daily_returns = calculate_daily_returns(portfolio_values)

    if len(daily_returns) < 2:
        raise ValueError("Need at least two return observations.")

    returns_list = [daily_returns[d] for d in sorted(daily_returns.keys())]
    return calculate_volatility(returns_list, periods_in_year)

calculate_volatility_ratio(returns)

Calculate the ratio of downside to upside volatility.

A ratio greater than 1 indicates more variability in losses than gains. A ratio less than 1 indicates more variability in gains than losses.

Parameters:

Name	Type	Description	Default
returns	list[float]	List of daily returns as decimals (e.g., 0.002 = 0.2%).	required

Returns:

Type	Description
float | None	The volatility ratio (downside/upside), or None if insufficient data.

Source code in src/athena/metrics/volatility.py

def calculate_volatility_ratio(
    returns: list[float]
) -> float | None:
    """
    Calculate the ratio of downside to upside volatility.

    A ratio greater than 1 indicates more variability in losses than gains.
    A ratio less than 1 indicates more variability in gains than losses.

    Args:
        returns: List of daily returns as decimals (e.g., 0.002 = 0.2%).

    Returns:
        The volatility ratio (downside/upside), or None if insufficient data.
    """
    positive_returns = [r for r in returns if r > 0]
    negative_returns = [r for r in returns if r < 0]

    if len(positive_returns) < 2 or len(negative_returns) < 2:
        return None

    upside_std = np.array(positive_returns, dtype=float).std(ddof=1)
    downside_std = np.array(negative_returns, dtype=float).std(ddof=1)

    if np.isclose(upside_std, 0.0):
        return None

    return float(downside_std / upside_std)

calculate_volatility_statistics(portfolio, target_currency, start_date=None, end_date=None, periods_in_year=365)

Calculate comprehensive volatility statistics for a portfolio.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions and settings.	required
target_currency	Currency	The currency to convert all values into.	required
start_date	datetime | None	The start date for the calculation (inclusive). If None, uses the earliest transaction date.	None
end_date	datetime | None	The end date for the calculation (inclusive). If None, uses the latest transaction date.	None
periods_in_year	int	Trading periods in a year (365 for daily).	365

Returns:

Type	Description
dict[str, float | None]	Dictionary containing:
dict[str, float | None]	daily_volatility: Daily standard deviation of returns
dict[str, float | None]	annual_volatility: Annualized standard deviation
dict[str, float | None]	daily_upside_volatility: Daily std of positive returns (or None)
dict[str, float | None]	annual_upside_volatility: Annualized std of positive returns (or None)
dict[str, float | None]	daily_downside_volatility: Daily std of negative returns (or None)
dict[str, float | None]	annual_downside_volatility: Annualized std of negative returns (or None)
dict[str, float | None]	volatility_ratio: Downside/upside volatility ratio (or None)

Source code in src/athena/metrics/volatility.py

def calculate_volatility_statistics(
    portfolio: Portfolio,
    target_currency: Currency,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    periods_in_year: int = 365
) -> dict[str, float | None]:
    """
    Calculate comprehensive volatility statistics for a portfolio.

    Args:
        portfolio: Portfolio object containing transactions and settings.
        target_currency: The currency to convert all values into.
        start_date: The start date for the calculation (inclusive).
                    If None, uses the earliest transaction date.
        end_date: The end date for the calculation (inclusive).
                  If None, uses the latest transaction date.
        periods_in_year: Trading periods in a year (365 for daily).

    Returns:
        Dictionary containing:
        - daily_volatility: Daily standard deviation of returns
        - annual_volatility: Annualized standard deviation
        - daily_upside_volatility: Daily std of positive returns (or None)
        - annual_upside_volatility: Annualized std of positive returns (or None)
        - daily_downside_volatility: Daily std of negative returns (or None)
        - annual_downside_volatility: Annualized std of negative returns (or None)
        - volatility_ratio: Downside/upside volatility ratio (or None)
    """
    # Get daily portfolio values
    portfolio_values = calculate_portfolio_value_by_day(
        portfolio,
        target_currency,
        start_date,
        end_date
    )

    if len(portfolio_values) < 2:
        return {
            "daily_volatility": None,
            "annual_volatility": None,
            "daily_upside_volatility": None,
            "annual_upside_volatility": None,
            "daily_downside_volatility": None,
            "annual_downside_volatility": None,
            "volatility_ratio": None,
        }

    daily_returns = calculate_daily_returns(portfolio_values)
    returns_list = [daily_returns[d] for d in sorted(daily_returns.keys())]

    # Calculate overall volatility
    try:
        daily_vol, annual_vol = calculate_volatility(returns_list, periods_in_year)
    except ValueError:
        daily_vol, annual_vol = None, None

    # Calculate upside volatility
    try:
        daily_up_vol, annual_up_vol = calculate_upside_volatility(returns_list, periods_in_year)
    except ValueError:
        daily_up_vol, annual_up_vol = None, None

    # Calculate downside volatility
    try:
        daily_down_vol, annual_down_vol = calculate_downside_volatility(returns_list, periods_in_year)
    except ValueError:
        daily_down_vol, annual_down_vol = None, None

    # Calculate volatility ratio
    vol_ratio = calculate_volatility_ratio(returns_list)

    return {
        "daily_volatility": daily_vol,
        "annual_volatility": annual_vol,
        "daily_upside_volatility": daily_up_vol,
        "annual_upside_volatility": annual_up_vol,
        "daily_downside_volatility": daily_down_vol,
        "annual_downside_volatility": annual_down_vol,
        "volatility_ratio": vol_ratio,
    }

calculate_win_rate_all_positions(portfolio, as_of=None)

Calculate win rate based on all positions (open and closed).

For closed positions, uses realized gain/loss. For open positions, uses unrealized gain/loss based on current market value.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions.	required
as_of	datetime | None	The datetime to evaluate positions at. If None, uses the latest transaction date.	None

Returns:

Type	Description
WinRateResult	WinRateResult containing win rate statistics for all positions.

Source code in src/athena/metrics/win_rate.py

def calculate_win_rate_all_positions(
    portfolio: Portfolio,
    as_of: datetime | None = None
) -> WinRateResult:
    """
    Calculate win rate based on all positions (open and closed).

    For closed positions, uses realized gain/loss.
    For open positions, uses unrealized gain/loss based on current market value.

    Args:
        portfolio: Portfolio object containing transactions.
        as_of: The datetime to evaluate positions at.
               If None, uses the latest transaction date.

    Returns:
        WinRateResult containing win rate statistics for all positions.
    """
    if as_of is None:
        if not portfolio.transactions:
            return WinRateResult(
                win_rate=0.0,
                total_positions=0,
                winning_positions=0,
                losing_positions=0,
                breakeven_positions=0,
                total_gain_loss=Decimal("0"),
                average_win=None,
                average_loss=None,
                win_loss_ratio=None
            )
        sorted_txns = sorted(portfolio.transactions, key=lambda t: t.transaction_datetime)
        as_of = sorted_txns[-1].transaction_datetime

    # Get closed positions
    closed_positions = get_closed_positions(portfolio, as_of)

    # Get open positions
    open_positions = get_positions(as_of, portfolio)

    # Combine gains/losses
    all_gains_losses: list[Decimal] = []

    # Add closed position gains/losses
    for pos in closed_positions:
        all_gains_losses.append(pos.realized_gain_loss)

    # Add open position gains/losses (unrealized)
    for pos in open_positions:
        if pos.gain_loss is not None:
            all_gains_losses.append(pos.gain_loss)

    if not all_gains_losses:
        return WinRateResult(
            win_rate=0.0,
            total_positions=0,
            winning_positions=0,
            losing_positions=0,
            breakeven_positions=0,
            total_gain_loss=Decimal("0"),
            average_win=None,
            average_loss=None,
            win_loss_ratio=None
        )

    winning = [gl for gl in all_gains_losses if gl > 0]
    losing = [gl for gl in all_gains_losses if gl < 0]
    breakeven = [gl for gl in all_gains_losses if gl == 0]

    total_positions = len(all_gains_losses)
    winning_positions = len(winning)
    losing_positions = len(losing)
    breakeven_positions = len(breakeven)

    win_rate = winning_positions / total_positions if total_positions > 0 else 0.0

    total_gain_loss = sum(all_gains_losses)

    average_win = (
        sum(winning) / winning_positions
        if winning_positions > 0 else None
    )

    average_loss = (
        sum(losing) / losing_positions
        if losing_positions > 0 else None
    )

    win_loss_ratio = (
        float(abs(average_win / average_loss))
        if average_win is not None and average_loss is not None and average_loss != 0
        else None
    )

    return WinRateResult(
        win_rate=win_rate,
        total_positions=total_positions,
        winning_positions=winning_positions,
        losing_positions=losing_positions,
        breakeven_positions=breakeven_positions,
        total_gain_loss=total_gain_loss,
        average_win=average_win,
        average_loss=average_loss,
        win_loss_ratio=win_loss_ratio
    )

calculate_win_rate_by_symbol(portfolio, as_of=None, include_open=False)

Calculate win rate statistics grouped by symbol.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions.	required
as_of	datetime | None	Only consider transactions up to this datetime. If None, considers all transactions.	None
include_open	bool	If True, includes unrealized gains from open positions.	False

Returns:

Type	Description
dict[str, WinRateResult]	Dictionary mapping symbol to WinRateResult.

Source code in src/athena/metrics/win_rate.py

def calculate_win_rate_by_symbol(
    portfolio: Portfolio,
    as_of: datetime | None = None,
    include_open: bool = False
) -> dict[str, WinRateResult]:
    """
    Calculate win rate statistics grouped by symbol.

    Args:
        portfolio: Portfolio object containing transactions.
        as_of: Only consider transactions up to this datetime.
               If None, considers all transactions.
        include_open: If True, includes unrealized gains from open positions.

    Returns:
        Dictionary mapping symbol to WinRateResult.
    """
    if as_of is None and portfolio.transactions:
        sorted_txns = sorted(portfolio.transactions, key=lambda t: t.transaction_datetime)
        as_of = sorted_txns[-1].transaction_datetime

    closed_positions = get_closed_positions(portfolio, as_of)

    # Group closed positions by symbol
    by_symbol: dict[str, list[Decimal]] = {}
    for pos in closed_positions:
        if pos.symbol not in by_symbol:
            by_symbol[pos.symbol] = []
        by_symbol[pos.symbol].append(pos.realized_gain_loss)

    # Add open positions if requested
    if include_open and as_of is not None:
        open_positions = get_positions(as_of, portfolio)
        for pos in open_positions:
            if pos.gain_loss is not None:
                if pos.symbol not in by_symbol:
                    by_symbol[pos.symbol] = []
                by_symbol[pos.symbol].append(pos.gain_loss)

    # Calculate win rate for each symbol
    results: dict[str, WinRateResult] = {}
    for symbol, gains_losses in by_symbol.items():
        if not gains_losses:
            continue

        winning = [gl for gl in gains_losses if gl > 0]
        losing = [gl for gl in gains_losses if gl < 0]
        breakeven = [gl for gl in gains_losses if gl == 0]

        total_positions = len(gains_losses)
        winning_positions = len(winning)
        losing_positions = len(losing)
        breakeven_positions = len(breakeven)

        win_rate = winning_positions / total_positions if total_positions > 0 else 0.0

        total_gain_loss = sum(gains_losses)

        average_win = (
            sum(winning) / winning_positions
            if winning_positions > 0 else None
        )

        average_loss = (
            sum(losing) / losing_positions
            if losing_positions > 0 else None
        )

        win_loss_ratio = (
            float(abs(average_win / average_loss))
            if average_win is not None and average_loss is not None and average_loss != 0
            else None
        )

        results[symbol] = WinRateResult(
            win_rate=win_rate,
            total_positions=total_positions,
            winning_positions=winning_positions,
            losing_positions=losing_positions,
            breakeven_positions=breakeven_positions,
            total_gain_loss=total_gain_loss,
            average_win=average_win,
            average_loss=average_loss,
            win_loss_ratio=win_loss_ratio
        )

    return results

calculate_win_rate_closed(portfolio, as_of=None)

Calculate win rate based on closed positions only.

A winning position is one where the sell price is higher than the buy price. Uses FIFO matching to pair buys with sells.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions.	required
as_of	datetime | None	Only consider transactions up to this datetime. If None, considers all transactions.	None

Returns:

Type	Description
WinRateResult	WinRateResult containing win rate statistics for closed positions.

Source code in src/athena/metrics/win_rate.py

def calculate_win_rate_closed(
    portfolio: Portfolio,
    as_of: datetime | None = None
) -> WinRateResult:
    """
    Calculate win rate based on closed positions only.

    A winning position is one where the sell price is higher than the buy price.
    Uses FIFO matching to pair buys with sells.

    Args:
        portfolio: Portfolio object containing transactions.
        as_of: Only consider transactions up to this datetime.
               If None, considers all transactions.

    Returns:
        WinRateResult containing win rate statistics for closed positions.
    """
    closed_positions = get_closed_positions(portfolio, as_of)

    if not closed_positions:
        return WinRateResult(
            win_rate=0.0,
            total_positions=0,
            winning_positions=0,
            losing_positions=0,
            breakeven_positions=0,
            total_gain_loss=Decimal("0"),
            average_win=None,
            average_loss=None,
            win_loss_ratio=None
        )

    winning = [p for p in closed_positions if p.realized_gain_loss > 0]
    losing = [p for p in closed_positions if p.realized_gain_loss < 0]
    breakeven = [p for p in closed_positions if p.realized_gain_loss == 0]

    total_positions = len(closed_positions)
    winning_positions = len(winning)
    losing_positions = len(losing)
    breakeven_positions = len(breakeven)

    win_rate = winning_positions / total_positions if total_positions > 0 else 0.0

    total_gain_loss = sum(p.realized_gain_loss for p in closed_positions)

    average_win = (
        sum(p.realized_gain_loss for p in winning) / winning_positions
        if winning_positions > 0 else None
    )

    average_loss = (
        sum(p.realized_gain_loss for p in losing) / losing_positions
        if losing_positions > 0 else None
    )

    win_loss_ratio = (
        float(abs(average_win / average_loss))
        if average_win is not None and average_loss is not None and average_loss != 0
        else None
    )

    return WinRateResult(
        win_rate=win_rate,
        total_positions=total_positions,
        winning_positions=winning_positions,
        losing_positions=losing_positions,
        breakeven_positions=breakeven_positions,
        total_gain_loss=total_gain_loss,
        average_win=average_win,
        average_loss=average_loss,
        win_loss_ratio=win_loss_ratio
    )

get_closed_positions(portfolio, as_of=None)

Get all closed positions from a portfolio.

A closed position is created when shares are sold. This uses FIFO (First In, First Out) matching to pair buys with sells.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	Portfolio object containing transactions.	required
as_of	datetime | None	Only consider transactions up to this datetime. If None, considers all transactions.	None

Returns:

Type	Description
list[ClosedPosition]	List of ClosedPosition objects representing realized trades.

Source code in src/athena/metrics/win_rate.py

def get_closed_positions(
    portfolio: Portfolio,
    as_of: datetime | None = None
) -> list[ClosedPosition]:
    """
    Get all closed positions from a portfolio.

    A closed position is created when shares are sold. This uses FIFO
    (First In, First Out) matching to pair buys with sells.

    Args:
        portfolio: Portfolio object containing transactions.
        as_of: Only consider transactions up to this datetime.
               If None, considers all transactions.

    Returns:
        List of ClosedPosition objects representing realized trades.
    """
    # Track open lots for each symbol using FIFO
    # Each lot is (quantity, price, date, currency)
    open_lots: dict[str, list[tuple[Decimal, Decimal, datetime, Currency]]] = {}
    closed_positions: list[ClosedPosition] = []

    sorted_transactions = sorted(
        portfolio.transactions,
        key=lambda t: t.transaction_datetime
    )

    for txn in sorted_transactions:
        if as_of is not None and txn.transaction_datetime > as_of:
            break

        symbol = txn.symbol
        quantity = txn.quantity
        price = Decimal(str(txn.price))
        currency = txn.currency

        if txn.transaction_type == TransactionType.BUY:
            if symbol not in open_lots:
                open_lots[symbol] = []
            open_lots[symbol].append((quantity, price, txn.transaction_datetime, currency))

        elif txn.transaction_type == TransactionType.SELL:
            if symbol not in open_lots or not open_lots[symbol]:
                continue

            remaining_to_sell = quantity
            sell_price = price
            sell_date = txn.transaction_datetime

            while remaining_to_sell > 0 and open_lots[symbol]:
                lot = open_lots[symbol][0]
                lot_qty, lot_price, lot_date, lot_currency = lot

                if lot_qty <= remaining_to_sell:
                    # Close entire lot
                    open_lots[symbol].pop(0)
                    closed_qty = lot_qty
                    remaining_to_sell -= lot_qty
                else:
                    # Partially close lot
                    open_lots[symbol][0] = (
                        lot_qty - remaining_to_sell,
                        lot_price,
                        lot_date,
                        lot_currency
                    )
                    closed_qty = remaining_to_sell
                    remaining_to_sell = Decimal("0")

                # Calculate realized gain/loss
                buy_value = closed_qty * lot_price
                sell_value = closed_qty * sell_price
                realized_gain_loss = sell_value - buy_value

                if buy_value != 0:
                    realized_gain_loss_percent = (realized_gain_loss / buy_value) * 100
                else:
                    realized_gain_loss_percent = Decimal("0")

                closed_positions.append(ClosedPosition(
                    symbol=symbol,
                    quantity=closed_qty,
                    buy_price=lot_price,
                    sell_price=sell_price,
                    buy_date=lot_date,
                    sell_date=sell_date,
                    currency=lot_currency,
                    realized_gain_loss=realized_gain_loss,
                    realized_gain_loss_percent=realized_gain_loss_percent
                ))

    return closed_positions