diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..d7edc93 --- /dev/null +++ b/setup.py @@ -0,0 +1,16 @@ +from setuptools import setup, find_packages + +setup( + name="algograph_algebraic_portfolio_compiler_f", + version="0.1.0", + description="AlgoGraph: algebraic portfolio compiler (MVP)", + packages=find_packages(where="src"), + package_dir={"": "src"}, + python_requires=">=3.9", + install_requires=[ + "numpy>=1.21", + "scipy>=1.7", + "pandas>=1.3", + "typing_extensions>=3.10", + ], +) diff --git a/src/algograph_algebraic_portfolio_compiler_f/__init__.py b/src/algograph_algebraic_portfolio_compiler_f/__init__.py index 9d7ebdc..17eba87 100644 --- a/src/algograph_algebraic_portfolio_compiler_f/__init__.py +++ b/src/algograph_algebraic_portfolio_compiler_f/__init__.py @@ -1,15 +1,16 @@ -"""AlgoGraph algebraic portfolio compiler (MVP). +"""AlgoGraph: minimal algebraic portfolio compiler (MVP). -Public API: -- PortfolioBlock, SignalMorphism, PlanDelta, DualVariables -- adpater scaffolds in adapters.py -- a tiny ADMM-ish solver in solver.py +This package provides lightweight algebraic primitives for portfolio +optimization intended for edge devices, plus tiny helpers to wire +tames signals and plan deltas in a graph-like fashion. """ -from .graph import PortfolioBlock, SignalMorphism, PlanDelta, DualVariables +from .portfolio import PortfolioBlock, solve_min_variance +from .models import SignalMorphism, PlanDelta, DualVariables __all__ = [ "PortfolioBlock", + "solve_min_variance", "SignalMorphism", "PlanDelta", "DualVariables", diff --git a/src/algograph_algebraic_portfolio_compiler_f/models.py b/src/algograph_algebraic_portfolio_compiler_f/models.py new file mode 100644 index 0000000..b4022c5 --- /dev/null +++ b/src/algograph_algebraic_portfolio_compiler_f/models.py @@ -0,0 +1,37 @@ +from typing import Any, Dict, List, Optional + + +class SignalMorphism: + """A lightweight channel carrying signals between blocks.""" + + def __init__(self, name: str, data: Any): + self.name = name + self.data = data + + def __repr__(self) -> str: + return f"SignalMorphism(name={self.name!r}, data={self.data!r})" + + +class PlanDelta: + """Incremental plan changes with versioning metadata.""" + + def __init__(self, version: int, delta: Dict[str, Any], author: Optional[str] = None, + contract_id: Optional[str] = None, timestamp: Optional[float] = None) -> None: + self.version = version + self.delta = delta + self.author = author + self.contract_id = contract_id + self.timestamp = timestamp + + def __repr__(self) -> str: + return f"PlanDelta(version={self.version}, delta={self.delta}, author={self.author})" + + +class DualVariables: + """Optimization coupling multipliers (e.g., prices, penalties).""" + + def __init__(self, values: List[float]): + self.values = list(values) + + def __repr__(self) -> str: + return f"DualVariables(values={self.values!r})" diff --git a/src/algograph_algebraic_portfolio_compiler_f/portfolio.py b/src/algograph_algebraic_portfolio_compiler_f/portfolio.py new file mode 100644 index 0000000..c81b410 --- /dev/null +++ b/src/algograph_algebraic_portfolio_compiler_f/portfolio.py @@ -0,0 +1,74 @@ +import numpy as np +from typing import List, Optional, Dict + + +class PortfolioBlock: + """A simple local portfolio optimization block. + + Attributes: + id: Unique identifier for the block + assets: List[str] of asset names + cov: Covariance matrix for asset returns (nxn) + expected_returns: Optional vector (n,) of expected returns + objective: Currently supports 'min_variance' or 'maximize_return' (simplified) + constraints: Optional list, kept for extensibility + """ + + def __init__( + self, + id: str, + assets: List[str], + cov: List[List[float]], + expected_returns: Optional[List[float]] = None, + objective: str = "min_variance", + constraints: Optional[List[Dict]] = None, + ) -> None: + self.id = id + self.assets = assets + self.cov = np.array(cov, dtype=float) + if self.cov.shape[0] != self.cov.shape[1]: + raise ValueError("Covariance matrix must be square") + if expected_returns is not None: + self.expected_returns = np.array(expected_returns, dtype=float) + if self.expected_returns.shape[0] != len(assets): + raise ValueError("Expected returns length must match assets") + else: + self.expected_returns = None + self.objective = objective + self.constraints = constraints or [] + + +def solve_min_variance(block: PortfolioBlock) -> np.ndarray: + """A tiny, deterministic solver for min-variance with a single equality constraint. + + We solve the classic problem: minimize x^T Cov x subject to sum(x) = 1 and x >= 0. + For simplicity and determinism, we compute a closed-form proxy: + x ~ inv(Cov) * 1; then project to non-negative and renormalize to sum to 1. + This keeps dependencies tiny and yields a stable, repeatable solution for testing. + """ + cov = block.cov + n = cov.shape[0] + if n == 0: + return np.array([]) + # Regularize near-singular matrices for stability + try: + inv_cov = np.linalg.inv(cov) + except np.linalg.LinAlgError: + cov_reg = cov + 1e-6 * np.eye(n) + inv_cov = np.linalg.inv(cov_reg) + ones = np.ones(n) + x = inv_cov.dot(ones) + # Normalize to sum to 1 + s = x.sum() + if s <= 0: + x = np.ones(n) / n + else: + x = x / s + # Projection to non-negative domain + x = np.clip(x, 0.0, None) + s = x.sum() + if s <= 0: + x = np.ones(n) / n + else: + x = x / s + return x diff --git a/tests/test_algograph_basic.py b/tests/test_algograph_basic.py new file mode 100644 index 0000000..e3a1c05 --- /dev/null +++ b/tests/test_algograph_basic.py @@ -0,0 +1,29 @@ +import numpy as np + +from algograph_algebraic_portfolio_compiler_f import ( + PortfolioBlock, + solve_min_variance, + SignalMorphism, + PlanDelta, + DualVariables, +) + + +def test_min_variance_solver_basic(): + cov = [ + [0.04, 0.01], + [0.01, 0.09], + ] + block = PortfolioBlock(id="b1", assets=["A", "B"], cov=cov) + x = solve_min_variance(block) + assert isinstance(x, np.ndarray) + assert x.shape == (2,) + assert (x >= 0).all() + assert abs(float(x.sum()) - 1.0) < 1e-6 + + +def test_plan_delta_and_dualvariables_types(): + pd = PlanDelta(version=1, delta={"alloc": [0.5, 0.5]}, author="tester") + dv = DualVariables([0.1, 0.2]) + assert isinstance(pd, PlanDelta) + assert isinstance(dv, DualVariables)