Evaluate constrained historical variance

from pathlib import Path
import base64, gzip, hashlib, json, platform
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from IPython.display import display, HTML
from scipy.optimize import minimize
from datetime import datetime
from IPython.display import display as _display
def display(value):
    if isinstance(value,pd.DataFrame):
        table=value.to_html(max_rows=12,escape=True)
        style="<style>.vibeit-readable-table{max-width:100%;overflow-x:auto}.vibeit-readable-table table{width:max-content;max-width:none;border-collapse:collapse}.vibeit-readable-table td,.vibeit-readable-table th{white-space:nowrap!important;word-break:normal!important;overflow-wrap:normal!important}</style>"
        return _display(HTML(style+'<div class="vibeit-readable-table">'+table+'</div>'))
    return _display(value)
RECIPE_ID='ec05-portfolio-variance'
OUTPUT_DIR=Path.cwd()/"results"/RECIPE_ID
plt.rcParams.update({"figure.dpi":120,"font.size":10,"axes.spines.top":False,"axes.spines.right":False,
                      "axes.prop_cycle":plt.cycler(color=["#18785f","#416aa6","#bc6541","#9974a5"])})
print("Python:",platform.python_version(),"; NumPy:",np.__version__,"; Pandas:",pd.__version__)
print("Working folder:",Path.cwd())
print("Core data mode: embedded snapshot; no network requests")

Python: 3.13.13 ; NumPy: 2.5.3 ; Pandas: 3.0.6
Working folder: /Users/zhiluo/Developer/vibeit-cookbook/cookbook/downloads
Core data mode: embedded snapshot; no network requests

def load_snapshot(recipe_id, expected_sha256, snapshot_key):
    """Read embedded data from this notebook, including a renamed copy."""
    for path in sorted(Path.cwd().glob("*.ipynb")):
        try:
            notebook = json.loads(path.read_text(encoding="utf-8"))
            metadata = notebook.get("metadata", {}).get("vibeit_cookbook", {})
            blob = metadata.get("snapshots", {}).get(snapshot_key)
            if metadata.get("recipe_id") != recipe_id or not blob:
                continue
            if blob.get("sha256") != expected_sha256:
                continue
            raw = gzip.decompress(base64.b64decode(blob["gzip_base64"], validate=True))
            if hashlib.sha256(raw).hexdigest() != expected_sha256:
                raise ValueError("Embedded snapshot checksum mismatch: " + snapshot_key)
            return json.loads(raw)
        except (OSError, json.JSONDecodeError):
            continue
    raise FileNotFoundError(
        "Save/open the complete .ipynb in the active working folder. "
        "The embedded snapshot could not be located for " + recipe_id)

SNAPSHOT_HASHES={'ecb': '7afd344000313ec2b4f6fb2548894e2a4817dc3687c14f8de47937ebf7bba0ad'}
def snapshot(key):
    return load_snapshot(RECIPE_ID,SNAPSHOT_HASHES[key],key)
print("Embedded snapshots:",list(SNAPSHOT_HASHES))

Embedded snapshots: ['ecb']

def minimum_variance_weights(covariance):
    """Long-only fully invested variance minimization; no expected-return prediction."""
    covariance=np.asarray(covariance,dtype=float)
    if covariance.ndim!=2 or covariance.shape[0]!=covariance.shape[1] or len(covariance)<2 or not np.isfinite(covariance).all():
        raise ValueError('Need a finite square covariance matrix')
    if not np.allclose(covariance,covariance.T) or np.linalg.eigvalsh(covariance).min()<-1e-10:
        raise ValueError('Covariance must be symmetric positive semidefinite')
    n=len(covariance);initial=np.full(n,1/n)
    result=minimize(lambda w:1e4*float(w@covariance@w),initial,jac=lambda w:2e4*covariance@w,
        method='SLSQP',bounds=[(0,1)]*n,constraints={'type':'eq','fun':lambda w:w.sum()-1,'jac':lambda w:np.ones(n)},
        options={'ftol':1e-12,'maxiter':500})
    if not result.success:raise RuntimeError(result.message)
    weights=result.x
    if np.any(weights<-1e-8) or not np.isclose(weights.sum(),1):raise RuntimeError('Optimizer violated weight constraints')
    return weights

print("Method ready: minimum_variance_weights")

Method ready: minimum_variance_weights

def iso_datetimes(values):
    """Parse source ISO dates with the standard library; keep tables as ISO strings."""
    return [datetime.fromisoformat(str(value)) for value in values]

print("Method ready: iso_datetimes")

Method ready: iso_datetimes

ecb=snapshot("ecb");rates=pd.DataFrame(ecb["rows"])
rates["date"]=rates["Date"].astype(str);rates=rates.set_index("date").sort_index()
currencies=ecb["currency_order"];original_dates=len(rates);rates=rates[currencies].dropna()
assert rates.index.is_unique and rates.index.is_monotonic_increasing and (rates>0).all().all()
prices=1/rates;returns=prices.pct_change(fill_method=None).iloc[1:]
print(ecb["source_notice"])
print("Quote:",ecb["quote"],"; common complete dates:",len(rates),"of",original_dates)
display(rates.head(6))
training=np.array([value.year<=2022 for value in iso_datetimes(returns.index)]);test=~training
assert returns.index[training].max()<returns.index[test].min()
covariance=returns.loc[training].cov().to_numpy();weights=minimum_variance_weights(covariance)
equal=np.full(4,.25);annualization=252
display(pd.DataFrame({"currency":currencies,"minimum_variance_weight":weights,"equal_weight":equal}))
fig,ax=plt.subplots(figsize=(7,4));positions=np.arange(4)
ax.bar(positions-.18,100*weights,width=.36,label="Training minimum variance");ax.bar(positions+.18,100*equal,width=.36,label="Equal")
ax.set_xticks(positions,currencies);ax.set(ylabel="Hypothetical weight (%)",title="Training-only constrained weights");ax.legend();plt.tight_layout();plt.show()

Source: European Central Bank. The original reference rates are available free from the ECB. Course returns, inversions and risk estimates are derived calculations, not ECB statistics.
Quote: Foreign currency units per 1 EUR ; common complete dates: 1538 of 1538

	USD	GBP	JPY	CNY
date
2019-01-02	1.1397	0.90165	124.28	7.8165
2019-01-03	1.1348	0.90312	122.21	7.8019
2019-01-04	1.1403	0.89988	123.20	7.8280
2019-01-07	1.1445	0.89720	123.90	7.8421
2019-01-08	1.1440	0.89743	124.46	7.8405
2019-01-09	1.1455	0.89913	124.70	7.8226

	currency	minimum_variance_weight	equal_weight
0	USD	0.118891	0.25
1	GBP	0.376475	0.25
2	JPY	0.210420	0.25
3	CNY	0.294214	0.25

series=pd.DataFrame({"minimum_variance":returns.to_numpy()@weights,"equal_weight":returns.to_numpy()@equal},index=returns.index)
volatility=pd.DataFrame({"training":[100*series.loc[training,c].std(ddof=1)*np.sqrt(annualization) for c in series],"held_out":[100*series.loc[test,c].std(ddof=1)*np.sqrt(annualization) for c in series]},index=series.columns)
fig,ax=plt.subplots(figsize=(8,4));volatility.plot.bar(ax=ax)
ax.set(ylabel="Annualized historical volatility (%)",xlabel="Fixed-weight strategy",title="Train/test historical comparison");plt.xticks(rotation=0);plt.tight_layout();plt.show()

growth=100*(1+series.loc[test]).cumprod()
fig,ax=plt.subplots(figsize=(9,4))
for code in growth:ax.plot(iso_datetimes(growth.index),growth[code],label=code)
ax.legend()
ax.set(xlabel="Held-out reference date",ylabel="Hypothetical price-only index (start=100)",title="Fixed training weights in held-out years");plt.tight_layout();plt.show()

assert np.isclose(weights.sum(),1) and np.all(weights>=-1e-8) and np.all(weights<=1+1e-8)
assert np.allclose(covariance,covariance.T)
assert weights@covariance@weights<=equal@covariance@equal+1e-10
OUTPUT_DIR.mkdir(parents=True,exist_ok=True)
pd.DataFrame({"currency":currencies,"weight":weights}).to_csv(OUTPUT_DIR/"training-only-weights.csv",index=False)
series.to_csv(OUTPUT_DIR/"hypothetical-price-returns.csv");volatility.to_csv(OUTPUT_DIR/"historical-volatility.csv")
(OUTPUT_DIR/"assumptions.json").write_text(json.dumps({"training_years":[2019,2022],"held_out_years":[2023,2024],"denomination":"EUR","interest_and_carry_included":False,"costs_and_tax_included":False,"rebalancing":"daily fixed weights","annualization_observations":252},indent=2))
print("Constraint and training-objective checks passed; exports:",OUTPUT_DIR)

Constraint and training-objective checks passed; exports: /Users/zhiluo/Developer/vibeit-cookbook/cookbook/downloads/results/ec05-portfolio-variance

RUN_ONLINE_UPDATE=False
if RUN_ONLINE_UPDATE:
    import requests
    try:
        response=requests.get('https://data-api.ecb.europa.eu/service/data/EXR/D.USD.EUR.SP00.A?startPeriod=2024-01-01&endPeriod=2024-12-31&format=csvdata',timeout=20)
        response.raise_for_status()
        print("Source:",response.url)
        print(response.text[:700])
    except requests.RequestException as error:
        print("Online extension unavailable:",type(error).__name__,str(error)[:160])
else:
    print("Online extension skipped; embedded core data unchanged")

Online extension skipped; embedded core data unchanged