math-backend / constraints.py
engineportf's picture
Upload folder using huggingface_hub
558db1e verified
Raw
History Blame Contribute Delete
4.36 kB
import math
import pandas as pd
import numpy as np
import numpy.linalg as la
from config import Color, logger, hr
def check_and_fix_bounds(tickers, asset_min, asset_max, sector_limit, sector_map, macro=None, gross_leverage_cap=1.0, silent=False):
"""
Validates and automatically resolves conflicting portfolio constraints.
Integrates with HMM regime detection to dynamically disable leverage in high-volatility crashes.
"""
n = len(tickers)
safe_min = asset_min
adj_gross_cap = gross_leverage_cap
warns = []
# ─────────────────────────────────────────────
# REGIME DETECTION: Dynamic Risk Management
# ─────────────────────────────────────────────
is_high_vol = False
if macro and "hmm_regime" in macro:
is_high_vol = macro["hmm_regime"].get("is_high_vol", False)
if is_high_vol and (safe_min < 0 or adj_gross_cap > 1.0):
safe_min = max(0.0, safe_min) # Disable shorts
adj_gross_cap = 1.0 # Disable leverage
warns.append(f" {Color.YELLOW}⚠ HMM High Volatility Regime: Short selling and leverage dynamically disabled to protect capital.{Color.RESET}")
# ─────────────────────────────────────────────
# STANDARD BOUNDS LOGIC
# ─────────────────────────────────────────────
if safe_min < 0:
if n * asset_max < 1.0:
asset_max = math.ceil(100.0 / n) / 100.0
warns.append(f" {Color.YELLOW}⚠ Maximum weight per asset raised to {asset_max:.0%} so the weights can add up to 100%.{Color.RESET}")
max_gl = n * max(asset_max, abs(safe_min))
if max_gl > 3.0:
warns.append(f" {Color.YELLOW}⚠ Your settings allow up to {max_gl:.1f}Γ— gross leverage β€” make sure this is intentional.{Color.RESET}")
else:
if safe_min * n > 0.999:
safe_min = 0.999 / n
warns.append(f" {Color.YELLOW}⚠ Minimum weight reduced to {safe_min:.2%} so all weights can still sum to 100%.{Color.RESET}")
sector_groups = {}
for t in tickers:
sector = sector_map.get(t, "Other")
if sector not in sector_groups:
sector_groups[sector] = []
sector_groups[sector].append(t)
for sector, members in sector_groups.items():
if sector == "Other":
continue
if safe_min * len(members) > sector_limit:
reduced = (sector_limit / len(members)) * 0.95
safe_min = min(safe_min, reduced)
warns.append(f" {Color.YELLOW}⚠ '{sector}' sector: minimum weight Γ— {len(members)} tickers would exceed the {sector_limit:.0%} sector cap. Minimum reduced to {reduced:.2%}.{Color.RESET}")
if safe_min >= asset_max:
safe_min = 0.0
warns.append(f" {Color.YELLOW}⚠ Minimum weight was β‰₯ maximum weight β€” reset minimum to 0 to avoid an impossible constraint.{Color.RESET}")
present_sectors = set(sector_map.get(t, "Other") for t in tickers)
num_sectors = max(1, len(present_sectors))
if num_sectors * sector_limit < 0.99:
old_lim = sector_limit
sector_limit = min(1.0, (1.0 / num_sectors) + 0.02)
warns.append(f" {Color.YELLOW}⚠ Sector limit mathematically forced from {old_lim:.0%} to {sector_limit:.0%} to ensure 100% portfolio coverage.{Color.RESET}")
if warns and not silent:
hr("─", 65, Color.YELLOW)
print(f"{Color.BOLD}{Color.YELLOW} Constraint Adjustments{Color.RESET}")
for w in warns:
print(w)
hr("─", 65, Color.YELLOW)
return safe_min, asset_max, adj_gross_cap, sector_limit
def make_nearest_psd(matrix):
"""Finds the nearest positive-definite matrix to input."""
sym_matrix = (matrix + matrix.T) / 2
eigval, eigvec = np.linalg.eigh(sym_matrix)
eigval[eigval < 0] = 1e-8
return eigvec @ np.diag(eigval) @ eigvec.T