build(agent): new-agents#a6e6ec iteration

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",
+    ],
+)

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:
+This package provides lightweight algebraic primitives for portfolio
-- PortfolioBlock, SignalMorphism, PlanDelta, DualVariables
+optimization intended for edge devices, plus tiny helpers to wire
-- adpater scaffolds in adapters.py
+tames signals and plan deltas in a graph-like fashion.
-- a tiny ADMM-ish solver in solver.py
 """
 
-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",

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})"

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

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)