Spaces:
Sleeping
Sleeping
| 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 | |