Optimization - Robust mean variance

import pandas as pd
import qcs
from qcs import (
    AllocationSnapshot,
    Constraint,
    Context,
    OptimizerConstraint,
    PositionConstraint,
)

positions = [
    {
        "ticker": "EQUITYEUR68",
        "allocation": 100.0,
        "long_short": "long_only",
        "locked": False,
        "turnover": 0.8,
        "expected_return": 0.01,
    },
    {
        "ticker": "EQUITYEUR70",
        "allocation": 200.0,
        "long_short": "long_only",
        "locked": False,
        "turnover": 0.8,
        "expected_return": 0.05,
    },
    {
        "ticker": "GBP",
        "allocation": 200.0,
        "long_short": "long_only",
        "locked": False,
        "turnover": 0.8,
        "expected_return": 0.005,
    },
]
# Note: long_short, locked and turnover properties might be ommited and later handled by apply_position_constraint

snapshot = AllocationSnapshot(currency="EUR", positions=positions)

# Create context
assets = qcs.import_assets_from_csv("examples/assets.csv")
histories = qcs.import_histories_from_csv(
    "examples/histories.csv", sep=",", date_format="%Y-%m-%d"
)
context = Context(
    date="2023-03-31",
    horizon="1m",
    local_db={"assets": assets, "histories": histories},
)


# Calculate historical returns (overwrite expected_return property for each position)
# snapshot = snapshot.get_historical_returns(context, window="1y")

# Calculate implied performance (overwrite expected_return property for each position)
# For this example we need to add the pricing_context and reference_market to the snapshot
pricing_context = {"risk_free_rate": 0.03}
reference_market = {
    "currency": "EUR",
    "positions": [
        {"ticker": "EQUITYEUR68", "allocation": 50},
        {"ticker": "EQUITYEUR70", "allocation": 50},
    ],
}
snapshot = AllocationSnapshot(
    currency="EUR",
    positions=positions,
    pricing_context=pricing_context,
    reference_market=reference_market,
)
snapshot = snapshot.get_market_implied_returns(
    context, implied_perf_exponent=3
)


# Apply other position constraints
snapshot = snapshot.apply_position_constraints(
    context,
    [
        PositionConstraint(
            type="long_short", filter=lambda asset: asset.code == "EUR"
        ),
        # Limit turnover to 50% for individual assets
        PositionConstraint(
            type="turnover", filter=lambda _asset: True, value=0.5
        ),
    ],
)

# Create portfolio constraints
constraints = [
    Constraint(
        type="target",
        value=1.0,
        filter=lambda _asset: True,
        label="Sum of weights = 1",
    ),
    Constraint(
        type="limit_max",
        value=0.5,
        filter=lambda asset: asset.type == "EQT",
        label="Equities < 50%",
    ),
]
portfolio_constraints = snapshot.build_portfolio_constraints(
    context, constraints
)

# Check if portfolio constraints are met for current snapshot
constraints_met, breached_constraints = snapshot.check_portfolio_constraints(
    portfolio_constraints, context
)
if constraints_met:
    print("All portfolio constraints are met for the given snapshot")
else:
    print(f"Some portfolio constraints are not met: {breached_constraints}")

# Check if portfolio constraints are feasible before running optimization
if not snapshot.are_constraints_feasible(
    context, portfolio_constraints=portfolio_constraints
):
    raise ValueError("Portfolio constraints are not feasible")

optimizer_constraints = [
    OptimizerConstraint(constraint_type="std", target=0.06)
]

# Provide other inputs
pivot_fields = ["type", "ticker"]
fields = ["\\\\div(MAV,MAV#)", "MAV", "STD%", "STD.BM%"]


print("\nOriginal portfolio risks:")
df_risks = snapshot.pivot_view(context, pivot_fields).get_risks(fields)
print(df_risks)


print("\nOptimized portfolio positions:")
optimized_snapshot = snapshot.optimize(
    context=context,
    portfolio_constraints=portfolio_constraints,
    optimizer_constraints=optimizer_constraints,
    cost_function="robust_mean_variance",
)

data = [
    {
        "ticker": position.ticker,
        "size": position.size,
        "expected_return": position.expected_return,
    }
    for position in optimized_snapshot.positions
]
df_positions = pd.DataFrame(data)
print(df_positions)

print("\nOptimized portfolio risks:")
df_risks = optimized_snapshot.pivot_view(context, pivot_fields).get_risks(
    fields
)
print(df_risks)