community
/
equicompiler-algebraic-portfolio-dsl-to-
34
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

build(agent): new-agents-4#58ba63 iteration

This commit is contained in:
agent-58ba63c88b4c9625 2026-04-22 17:24:14 +02:00
parent dc454c9000
commit 48c1f47367
10 changed files with 574 additions and 207 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
AGENTS.md
View File

@ -16,7 +16,7 @@ Architecture overview for the EquiCompiler MVP
- LocalProblem: a portfolio-level optimization task expressed in the DSL
- SharedVariables: signals and inter-node communication primitives (simplified in MVP)
- PlanDelta: incremental plan changes with versioning (simplified in MVP)
- Auditability: IR includes a version field; future work will embed cryptographic attestations
- Auditability: IR includes versioning, deterministic attestations, lineage digests, and a Graph-of-Contracts skeleton
- Repository structure (initial)
- AGENTS.md: this document
@ -24,15 +24,24 @@ Architecture overview for the EquiCompiler MVP
- pyproject.toml: packaging metadata
- equicompiler_algebraic_portfolio_dsl_to_/
- __init__.py
- core.py: DSL -> IR parser (minimal)
- cli.py: CLI entry point
- core.py: DSL -> EquiIR compiler, lineage, attestations, and execution-plan synthesis
- cli.py: CLI entry point with optional reference backtest
- delta_sync.py: deterministic IR diffing and plan delta summaries
- registry.py: in-process Graph-of-Contracts contract registry
- goc_registry.py: GoC skeleton generation helpers
- backends/python_backtester.py: deterministic Python reference backtester
- adapters/: starter feed and broker adapters
- tests/
- test_core.py: basic unit test for DSL parsing
- test_core.py: DSL parsing and core IR assertions
- test_execution_policy.py: execution policy parsing
- test_goC.py: GoC structure assertions
- test_version_override.py: DSL version override coverage
- test_compiler_pipeline.py: richer compiler, backtest, and delta-sync coverage
- test.sh: test runner that executes unit tests and builds the package
- Testing philosophy
- One basic test ensures the DSL is parsed into a structured IR.
- Tests should be deterministic and fast, enabling CI checks early.
- Cover parser normalization, deterministic backtesting, and delta-sync behavior.
- Roadmap (high level)
- Expand the DSL grammar and IR representation

168
README.md
View File

@ -1,91 +1,107 @@
# EquiCompiler: Algebraic Portfolio DSL to Verifiable Low-Latency Compiler (MVP)
# EquiCompiler
Overview
- This repository contains a minimal, production-ready MVP for a compiler stack that translates a domain-specific language for market strategies into a portable, verifiable execution graph. The MVP focuses on a small, well-formed core to enable safe extension by additional agents in the swarm.
EquiCompiler translates a compact portfolio DSL into a canonical EquiIR graph with deterministic attestations, registry-backed contracts, delta-sync metadata, and a reference Python backtest path.
What you get in this MVP
- A small DSL parser that reads a concise DSL and converts it into a canonical in-memory IR (EquiIR) represented as Python dicts.
- A minimal Python package equicompiler_algebraic_portfolio_dsl_to_ with a core module and a CLI entry point.
- A tiny test suite to validate DSL parsing and IR generation.
- A test runner (test.sh) that executes tests and builds the package to verify packaging metadata.
## What It Does
Usage
- DSL to IR (programmatic):
- Parses portfolio strategy declarations into a normalized IR.
- Captures assets, objective clauses, constraints, execution policy, risk budgets, and regulatory metadata.
- Builds a deterministic execution plan with backend targets for Python, C++, and WebAssembly.
- Emits digest-based attestations and a Graph-of-Contracts skeleton for auditability.
- Provides a reference Python backtester for offline evaluation and replay.
- Computes deterministic delta-sync summaries between IR revisions.
## DSL
Example:
```text
version: 0.3
assets: AAPL@0.6, MSFT@0.4
objectives: maximize_return; minimize_risk
constraints: max_drawdown=0.15, var=0.95, cvar=0.9
risk_budget: max_var=0.1, max_exposure=0.8
execution_policy: offline, delta_sync, rebalance=1d
regulatory: venue=us, suitability=true
```
Supported sections:
- `version`
- `assets`
- `objectives`
- `constraints`
- `execution_policy`
- `risk_budget`
- `market_microstructure`
- `regulatory`
## IR Surface
`parse_dsl_to_ir()` returns a dictionary with:
- canonical assets and optional weights
- objectives and constraints
- execution policy and risk budget
- compiler metadata and lineage digest
- execution plan with registry contracts
- attestations and GoC skeleton
- delta-sync metadata
## Reference Backtest
The Python backend accepts either provided price series or a deterministic synthesized scenario and returns:
- a portfolio curve
- total return
- volatility
- max drawdown
- VaR / CVaR estimates
- objective satisfaction
## CLI
```bash
python -m equicompiler_algebraic_portfolio_dsl_to_.cli path/to/strategy.dsl
python -m equicompiler_algebraic_portfolio_dsl_to_.cli path/to/strategy.dsl --backtest
```
## Programmatic Use
```python
from equicompiler_algebraic_portfolio_dsl_to_.core import parse_dsl_to_ir
dsl = "assets: AAPL, MSFT, GOOG\nobjectives: maximize_return\nconstraints: max_drawdown=0.2, var=0.95"
ir = parse_dsl_to_ir(dsl)
print(ir)
from equicompiler_algebraic_portfolio_dsl_to_.backends.python_backtester import run_backtest
- CLI (Python module):
python -m equicompiler_algebraic_portfolio_dsl_to_.cli path/to/dsl_file.txt
ir = parse_dsl_to_ir(dsl_text)
result = run_backtest(ir)
```
Verifiable DSL Overview
- The EquiCompiler MVP translates a mathematically precise portfolio DSL into a portable, verifiable IR (EquiIR).
- Each transformation step embeds lightweight cryptographic attestations (digest-based) to ensure auditability and replayability without leaking sensitive data.
- A Graph-of-Contracts (GoC) skeleton is attached to the IR to enable plug-and-play adapters for data feeds and brokers, while preserving versioned contracts and compatibility.
- The system supports offline-first evaluation with deterministic delta-sync, enabling local backtests and plan replay when connectivity resumes.
- Backends are designed for cross-runtime portability: Python backtester for offline tests, C++ for live trading, and WebAssembly for browser-based evaluation.
- The MVP keeps a minimal, stable surface area while providing clear extension points for future formal verification hooks and delta-sync workflows.
## Packaging
Example workflow
- Parse a DSL to IR (programmatic):
``from equicompiler_algebraic_portfolio_dsl_to_.core import parse_dsl_to_ir``
``dsl = "assets: AAPL, MSFT, GOOG\nobjectives: maximize_return\nconstraints: max_drawdown=0.2, var=0.95"``
``ir = parse_dsl_to_ir(dsl)``
``print(ir)``
- Use the CLI to validate and inspect the IR produced from a file:
``python -m equicompiler_algebraic_portfolio_dsl_to_.cli path/to/dsl_file.txt``
- Python package name: `equicompiler_algebraic_portfolio_dsl_to`
- Source package: `equicompiler_algebraic_portfolio_dsl_to_`
- Build metadata: `pyproject.toml`
- README used as the long-form package description
Roadmap (high level)
- Phase 0: DSL to IR, start backends (Python backtester, minimal C++ live skeleton) and a minimal GoC registry.
- Phase 1: Formal verification hooks and delta-sync for offline/online reconciliation.
- Phase 2: Cross-backend interoperability tests with toy adapters.
- Phase 3: Hardware-in-the-loop (HIL) testing with partitioned portfolios.
## Tests
Publishing readiness
- This repository is designed for packaging as a Python module named
``equicompiler_algebraic_portfolio_dsl_to`` with versioning in ``pyproject.toml``.
- The README now includes a marketing-style overview and usage examples to aid publication.
Run the local verification flow with:
Project structure
- AGENTS.md: architecture and testing guidelines for future agents
- README.md: this file
- pyproject.toml: packaging metadata
- equicompiler_algebraic_portfolio_dsl_to_/ (package)
- __init__.py
- core.py: DSL -> IR parser (minimal)
- cli.py: CLI entry point
- tests/test_core.py: small unit test for DSL parsing
- test.sh: test runner to validate test suite and packaging
```bash
bash test.sh
```
Development notes
- The MVP intentionally keeps dependencies minimal to ensure fast iterations and deterministic tests.
- When adding features, try to keep changes small and focused on a single goal.
- Ensure tests cover the new functionality and avoid sensitive data in tests.
That script installs the package in editable mode, runs the unit tests, and builds a source/wheel distribution.
Next steps
- Extend the DSL with richer constraints (VaR, VaR-CVaR, liquidity, latency) and ExecutionPolicy primitives.
- Integrate the GoC registry and build a canonical EquiIR representation with per-message metadata for replay/verification.
- Add a lightweight delta-sync coordinator and starter adapters for data feeds and brokers.
- Expand the test suite to exercise the new backtester and Graph-of-Contracts scaffolds.
- Improve packaging and docs to support publishing to a Python package index.
- Implement a more expressive DSL and a richer IR (EquiIR) representation.
- Add more tests for edge cases and simple integration tests for the CLI.
- Expand packaging metadata and README with a longer developer and user guide.
- Add a first-pass Graph-of-Contracts registry scaffold (GoC) and a minimal adapter registry.
- Documentation: GoC overview and how to plug in new adapters.
## Architecture Notes
Extensibility notes
- The repository now includes a small GoC registry module (equicompiler_algebraic_portfolio_dsl_to_/goc_registry.py) and a registry-aware GoC skeleton integrated into the IR formation flow. This provides a stable extension point for future adapters (data feeds, brokers) and verifiable contract graphs.
- You can register adapters via the GoCRegistry class and view a digest that helps ensure reproducible builds and auditability.
- `core.py`: DSL parsing, IR normalization, lineage, attestations, and plan synthesis
- `delta_sync.py`: deterministic IR diffing
- `registry.py`: in-process GoC contract registry
- `goc_registry.py`: GoC skeleton builder
- `backends/python_backtester.py`: deterministic reference backtester
- `adapters/`: starter adapter interfaces for feeds and brokers
- Extend the DSL with richer constraints (VaR, VaR-CVaR, liquidity, latency) and ExecutionPolicy primitives.
- Integrate the GoC registry and build a canonical EquiIR representation with per-message metadata for replay/verification.
- Add a lightweight delta-sync coordinator and starter adapters for data feeds and brokers.
- Expand the test suite to exercise the new backtester and Graph-of-Contracts scaffolds.
- Improve packaging and docs to support publishing to a Python package index.
- Implement a more expressive DSL and a richer IR (EquiIR) representation.
- Add more tests for edge cases and simple integration tests for the CLI.
- Expand packaging metadata and README with a longer developer and user guide.
## Status
This README intentionally stays light; the AGENTS.md contains the deeper architectural notes for the SWARM collaborators.
This repository now has a coherent compiler pipeline, but it is still a reference implementation rather than a full market execution platform.

6
equicompiler_algebraic_portfolio_dsl_to_/__init__.py
View File

@ -1,3 +1,5 @@
"""EquiCompiler Algebraic Portfolio DSL - package init"""
"""EquiCompiler Algebraic Portfolio DSL package."""
__all__ = ["core", "cli"]
from .core import compile_strategy, parse_dsl_to_ir, to_json
__all__ = ["compile_strategy", "parse_dsl_to_ir", "to_json"]

151
equicompiler_algebraic_portfolio_dsl_to_/backends/python_backtester.py
View File

@ -1,26 +1,143 @@
"""Lightweight Python backtester backend scaffold for EquiCompiler MVP."""
"""Deterministic Python reference backtester for EquiCompiler."""
from __future__ import annotations
from typing import Dict, Any
import hashlib
import json
import math
from statistics import mean, pstdev
from typing import Any, Dict, Iterable, List, Sequence
def run_backtest(ir: Dict[str, Any], data: Any = None) -> Dict[str, Any]:
"""Deterministic, simple backtest placeholder.
This function is a scaffold. It returns a deterministic, small summary based on
the IR contents (assets and objectives). It is not a real backtester, but it
provides a deterministic bridge for MVP tests and integration with the rest of
the stack.
"""
assets = ir.get("assets", [])
objectives = ir.get("objectives", [])
# Very naive placeholder: return a pretend annualized return depending on asset count
base_return = 0.05 * max(1, len(assets)) if assets else 0.0
"""Run a deterministic reference backtest against provided or synthesized prices."""
assets = list(ir.get("assets", []) or [])
prices = _normalize_prices(data, assets)
weights = _resolve_weights(ir, assets)
portfolio_curve = _build_portfolio_curve(prices, weights)
returns = _series_returns(portfolio_curve)
summary = {
"assets": assets,
"objective_met": any(o for o in objectives if isinstance(o, str) and o.lower().find("maximize") != -1),
"return": round(base_return, 4),
"trace": [{"step": "init", "value": base_return}],
"weights": weights,
"curve": portfolio_curve,
"metrics": _compute_metrics(portfolio_curve, returns),
"objective_met": _objective_met(ir, _compute_metrics(portfolio_curve, returns)),
"trace": [
{"step": "ingest", "assets": assets},
{"step": "simulate", "points": len(portfolio_curve)},
{"step": "measure", "metrics_digest": _digest({"curve": portfolio_curve})},
],
}
return summary
__all__ = ["run_backtest"]
def _normalize_prices(data: Any, assets: Sequence[str]) -> Dict[str, List[float]]:
if isinstance(data, dict) and data:
normalized: Dict[str, List[float]] = {}
for asset, series in data.items():
normalized[str(asset)] = [float(point) for point in series]
return normalized
return {asset: _synthetic_price_series(asset) for asset in assets}
def _synthetic_price_series(asset: str, periods: int = 16) -> List[float]:
seed = int(hashlib.sha256(asset.encode("utf-8")).hexdigest()[:8], 16)
base = 100.0 + (seed % 150)
series = [base]
for idx in range(1, periods):
direction = 1 if ((seed >> idx) & 1) else -1
magnitude = 0.0025 + ((seed >> (idx % 11)) % 7) * 0.0008
next_price = series[-1] * (1.0 + direction * magnitude)
series.append(round(max(1.0, next_price), 4))
return series
def _resolve_weights(ir: Dict[str, Any], assets: Sequence[str]) -> Dict[str, float]:
explicit = ir.get("asset_weights", {}) or {}
if explicit:
return {asset: float(explicit.get(asset, 0.0)) for asset in assets}
if not assets:
return {}
weight = 1.0 / len(assets)
return {asset: weight for asset in assets}
def _build_portfolio_curve(prices: Dict[str, List[float]], weights: Dict[str, float]) -> List[float]:
if not prices:
return [1.0]
length = min(len(series) for series in prices.values())
curve: List[float] = []
for idx in range(length):
value = 0.0
for asset, series in prices.items():
base = series[0] if series and series[0] else 1.0
point = series[idx]
value += weights.get(asset, 0.0) * (point / base)
curve.append(round(value, 6))
return curve
def _series_returns(curve: Sequence[float]) -> List[float]:
returns: List[float] = []
for idx in range(1, len(curve)):
prev = curve[idx - 1]
curr = curve[idx]
if prev == 0:
returns.append(0.0)
else:
returns.append((curr - prev) / prev)
return returns
def _compute_metrics(curve: Sequence[float], returns: Sequence[float]) -> Dict[str, float]:
start = curve[0] if curve else 1.0
end = curve[-1] if curve else 1.0
total_return = (end / start) - 1.0 if start else 0.0
volatility = pstdev(returns) if len(returns) > 1 else 0.0
max_drawdown = _max_drawdown(curve)
var_95, cvar_95 = _tail_risk(returns)
return {
"total_return": round(total_return, 6),
"volatility": round(volatility, 6),
"max_drawdown": round(max_drawdown, 6),
"var_95": round(var_95, 6),
"cvar_95": round(cvar_95, 6),
}
def _max_drawdown(curve: Sequence[float]) -> float:
peak = curve[0] if curve else 1.0
worst = 0.0
for value in curve:
peak = max(peak, value)
drawdown = (peak - value) / peak if peak else 0.0
worst = max(worst, drawdown)
return worst
def _tail_risk(returns: Sequence[float]) -> tuple[float, float]:
if not returns:
return 0.0, 0.0
ordered = sorted(returns)
tail_count = max(1, math.ceil(len(ordered) * 0.05))
tail = ordered[:tail_count]
return tail[-1], mean(tail)
def _objective_met(ir: Dict[str, Any], metrics: Dict[str, float]) -> bool:
objectives = [str(obj).lower() for obj in ir.get("objectives", []) or []]
if any("maximize" in objective for objective in objectives):
return metrics["total_return"] >= 0
if any("minimize" in objective for objective in objectives):
return metrics["max_drawdown"] <= float(ir.get("constraints", {}).get("max_drawdown", 1.0))
return bool(objectives)
def _digest(payload: Dict[str, Any]) -> str:
return hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()
def run_reference_backtest(ir: Dict[str, Any], data: Any = None) -> Dict[str, Any]:
return run_backtest(ir, data)
__all__ = ["run_backtest", "run_reference_backtest"]

24
equicompiler_algebraic_portfolio_dsl_to_/cli.py
View File

@ -1,20 +1,28 @@
"""CLI entry point for the MVP DSL -> IR translator."""
"""CLI entry point for the EquiCompiler DSL -> IR translator."""
from __future__ import annotations
import argparse
import json
import sys
from .backends.python_backtester import run_backtest
from .core import parse_dsl_to_ir
def main(argv=None):
if argv is None:
argv = sys.argv[1:]
if len(argv) != 1:
print("Usage: python -m equicompiler_algebraic_portfolio_dsl_to_.cli <dsl_file>")
sys.exit(2)
path = argv[0]
with open(path, "r", encoding="utf-8") as f:
parser = argparse.ArgumentParser(prog="equicompiler", description="Compile EquiCompiler DSL into EquiIR")
parser.add_argument("dsl_file", help="Path to a DSL source file")
parser.add_argument("--backtest", action="store_true", help="Run the Python reference backtester")
args = parser.parse_args(argv)
with open(args.dsl_file, "r", encoding="utf-8") as f:
dsl = f.read()
ir = parse_dsl_to_ir(dsl)
if args.backtest:
ir = dict(ir)
ir["reference_backtest"] = run_backtest(ir)
print(json.dumps(ir, indent=2, sort_keys=True))

321
equicompiler_algebraic_portfolio_dsl_to_/core.py
View File

@ -1,142 +1,271 @@
"""Core DSL -> IR translation for the EquiCompiler MVP."""
"""Core DSL -> IR translation for the EquiCompiler stack."""
from __future__ import annotations
import json
from typing import Dict, List
import copy
from datetime import datetime
import hashlib
import json
from datetime import datetime
from typing import Any, Dict, Iterable, List, Sequence, Tuple, cast
# Optional GoC registry integration (extensible adapters in the future)
try:
from .goc_registry import GoCRegistry, build_minimal_goC_skeleton
except Exception:
GoCRegistry = None # type: ignore
build_minimal_goC_skeleton = None # type: ignore
from .delta_sync import delta_sync
from .goc_registry import build_minimal_goC_skeleton
from .registry import GraphRegistry
COMPILER_NAME = "EquiCompiler"
DEFAULT_VERSION = "0.2"
SUPPORTED_BACKENDS = (
{"name": "python_backtest", "runtime": "python", "mode": "offline"},
{"name": "cpp_live", "runtime": "cpp", "mode": "live"},
{"name": "wasm_eval", "runtime": "wasm", "mode": "browser"},
)
def parse_dsl_to_ir(dsl: str) -> Dict[str, object]:
"""Parse a tiny, human-friendly DSL into a canonical IR dict.
"""Parse the strategy DSL into a canonical EquiIR dictionary."""
sections = _collect_sections(dsl)
version = _first_value(sections, "version") or DEFAULT_VERSION
assets, asset_weights = _parse_assets(_first_value(sections, "assets"))
objectives = _parse_list(_first_value(sections, "objectives"))
constraints = _parse_kv_pairs(_first_value(sections, "constraints"), coerce=False)
execution_policy = _parse_list(_first_value(sections, "execution_policy"))
risk_budget = _parse_kv_pairs(_first_value(sections, "risk_budget"), coerce=True)
market_microstructure = _parse_kv_pairs(_first_value(sections, "market_microstructure"), coerce=True)
regulatory = _parse_kv_pairs(_first_value(sections, "regulatory"), coerce=True)
Expected DSL syntax (very small subset for MVP):
assets: AAPL, MSFT, GOOG
objectives: maximize_return; minimize_risk
constraints: max_drawdown=0.2, var=0.95
metadata = {
"compiler": COMPILER_NAME,
"normalized_sections": sorted(sections.keys()),
"unknown_sections": {
key: values
for key, values in sections.items()
if key not in {
"assets",
"objectives",
"constraints",
"execution_policy",
"risk_budget",
"market_microstructure",
"regulatory",
"version",
}
},
}
Returns a dict with keys: assets (list[str]), objectives (list[str]), constraints (dict[str,str]), version (str)
"""
assets: List[str] = []
objectives: List[str] = []
constraints: Dict[str, str] = {}
execution_policy: List[str] = []
# Optional version override from the DSL
parsed_version: str | None = None
lines = [ln.strip() for ln in dsl.strip().splitlines() if ln.strip()]
for line in lines:
lower = line.lower()
if lower.startswith("assets:"):
rest = line[len("assets:"):].strip()
assets = [a.strip() for a in rest.split(",") if a.strip()]
elif lower.startswith("objectives:"):
rest = line[len("objectives:"):].strip()
parts = [p.strip() for p in rest.replace(";", ",").split(",") if p.strip()]
objectives = parts
elif lower.startswith("constraints:"):
rest = line[len("constraints:"):].strip()
pairs = [p.strip() for p in rest.split(",") if p.strip()]
for p in pairs:
if "=" in p:
k, v = [s.strip() for s in p.split("=", 1)]
constraints[k] = v
elif lower.startswith("execution_policy:"):
# Optional execution policy descriptor for offline/online orchestration
rest = line[len("execution_policy:"):].strip()
parts = [p.strip() for p in rest.replace(";", ",").split(",") if p.strip()]
execution_policy = parts
elif lower.startswith("version:"):
# Allow users to explicitly set the IR version via the DSL
rest = line[len("version:"):].strip()
if rest:
parsed_version = rest
ir = {
ir: Dict[str, object] = {
"compiler": {
"name": COMPILER_NAME,
"version": version,
"backend_targets": list(SUPPORTED_BACKENDS),
},
"version": version,
"assets": assets,
"asset_weights": asset_weights,
"objectives": objectives,
"constraints": constraints,
"execution_policy": execution_policy,
# Default remains 0.1 unless overridden by DSL via `version:` line
"version": parsed_version if parsed_version is not None else "0.1",
"risk_budget": risk_budget,
"market_microstructure": market_microstructure,
"regulatory": regulatory,
"metadata": metadata,
}
# Attach lightweight, deterministic attestations to support auditability.
ir = _attach_attestations(ir)
return ir
ir["plan"] = _build_execution_plan(ir)
ir["lineage"] = _build_lineage(ir)
return _attach_attestations(ir)
def to_json(ir: Dict[str, object]) -> str:
return json.dumps(ir, indent=2, sort_keys=True)
def _attach_attestations(ir: Dict[str, object]) -> Dict[str, object]:
"""Attach deterministic attestations to the IR for auditability.
def compile_strategy(dsl: str) -> Dict[str, object]:
"""Compatibility wrapper for callers that want a compiler-style entry point."""
return parse_dsl_to_ir(dsl)
This is a lightweight, deterministic placeholder for cryptographic attestations.
It computes a SHA-256 digest of the IR excluding the attestations field and
adds per-step attestations to the IR under the 'attestations' key.
"""
# Work on a shallow copy to avoid mutating input in unexpected ways
def _collect_sections(dsl: str) -> Dict[str, List[str]]:
sections: Dict[str, List[str]] = {}
for raw_line in dsl.splitlines():
line = raw_line.strip()
if not line or line.startswith("#"):
continue
if ":" not in line:
sections.setdefault("notes", []).append(line)
continue
key, value = line.split(":", 1)
key = key.strip().lower().replace("-", "_")
sections.setdefault(key, []).append(value.strip())
return sections
def _first_value(sections: Dict[str, List[str]], key: str) -> str:
values = sections.get(key, [])
return " ".join(v for v in values if v).strip()
def _parse_list(value: str) -> List[str]:
if not value:
return []
items = [part.strip() for part in value.replace(";", ",").split(",")]
return [item for item in items if item]
def _parse_assets(value: str) -> Tuple[List[str], Dict[str, float]]:
assets: List[str] = []
weights: Dict[str, float] = {}
for token in _parse_list(value):
symbol, weight = _split_symbol_weight(token)
assets.append(symbol)
if weight is not None:
weights[symbol] = weight
return assets, weights
def _split_symbol_weight(token: str) -> Tuple[str, float | None]:
for separator in ("@", ":"):
if separator in token:
symbol, raw_weight = token.split(separator, 1)
symbol = symbol.strip()
raw_weight = raw_weight.strip()
try:
return symbol, float(raw_weight)
except ValueError:
return symbol, None
return token.strip(), None
def _parse_kv_pairs(value: str, *, coerce: bool) -> Dict[str, object]:
parsed: Dict[str, object] = {}
if not value:
return parsed
for token in _parse_list(value):
if "=" not in token:
parsed[token] = True
continue
key, raw_value = token.split("=", 1)
parsed[key.strip()] = _coerce_scalar(raw_value.strip()) if coerce else raw_value.strip()
return parsed
def _coerce_scalar(value: str) -> object:
lowered = value.lower()
if lowered in {"true", "false"}:
return lowered == "true"
try:
if "." in value or "e" in lowered:
return float(value)
return int(value)
except ValueError:
return value
def _build_execution_plan(ir: Dict[str, object]) -> Dict[str, object]:
assets = cast(List[object], ir.get("assets", []))
objectives = cast(List[object], ir.get("objectives", []))
execution_policy = cast(List[object], ir.get("execution_policy", []))
registry = GraphRegistry()
registry.register_contract("equiir-core", str(ir.get("version", DEFAULT_VERSION)), "compiler")
registry.register_contract("python-backtest", "0.1", "python_backtester")
registry.register_contract("cpp-live", "0.1", "cpp_live_runtime")
registry.register_contract("wasm-eval", "0.1", "wasm_eval_runtime")
return {
"kind": "execution_plan",
"phases": [
{
"name": "normalize",
"actions": [
"canonicalize_assets",
"canonicalize_constraints",
"canonicalize_execution_policy",
],
},
{
"name": "verify",
"actions": [
"hash_ir",
"attach_attestations",
"build_graph_of_contracts",
],
},
{
"name": "dispatch",
"actions": [
"python_backtest",
"cpp_live",
"wasm_eval",
],
},
],
"targets": list(SUPPORTED_BACKENDS),
"contract_registry": registry.list_contracts(),
"summary": {
"asset_count": len(assets),
"objective_count": len(objectives),
"execution_policy_count": len(execution_policy),
},
}
def _build_lineage(ir: Dict[str, object]) -> Dict[str, object]:
payload = {
"version": ir.get("version"),
"assets": ir.get("assets", []),
"objectives": ir.get("objectives", []),
"constraints": ir.get("constraints", {}),
"execution_policy": ir.get("execution_policy", []),
"risk_budget": ir.get("risk_budget", {}),
}
digest = hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()
return {"digest": digest, "canonical": payload}
def _attach_attestations(ir: Dict[str, object]) -> Dict[str, object]:
"""Attach deterministic attestations to the IR for auditability."""
ir_copy = copy.deepcopy(ir)
base_for_digest = {k: v for k, v in ir_copy.items() if k != "attestations"}
digest = hashlib.sha256(json.dumps(base_for_digest, sort_keys=True).encode("utf-8")).hexdigest()
timestamp = datetime.utcnow().isoformat() + "Z"
attestations = [
ir_copy["attestations"] = [
{"step": "parse", "timestamp": timestamp, "digest": digest, "algorithm": "SHA-256"},
{"step": "normalize", "timestamp": timestamp, "digest": digest, "algorithm": "SHA-256"},
{"step": "verify", "timestamp": timestamp, "digest": digest, "algorithm": "SHA-256"},
]
ir_copy["attestations"] = attestations
# Attach a lightweight Graph-of-Contracts (GoC) skeleton to the IR
ir_copy["graph_of_contracts"] = _build_goC_skeleton(ir_copy, digest, timestamp)
ir_copy["delta_sync"] = {
"digest": digest,
"mode": "deterministic",
"summary": delta_sync({"version": ir_copy.get("version")}, ir_copy),
}
return ir_copy
def _build_goC_skeleton(ir_with_attestations: Dict[str, object], base_digest: str, base_timestamp: str) -> Dict[str, object]:
"""Create a minimal Graph-of-Contracts (GoC) skeleton for the IR.
The GoC is a placeholder scaffold intended for early interoperability between
adapters (data feeds, brokers) and the EquiIR. It includes a registry digest and
a generation timestamp to aid replay and auditability.
"""
# Attempt to compute a compact digest from core IR parts to seed the GoC metadata
"""Create a minimal Graph-of-Contracts skeleton for the IR."""
try:
base_for_goct = {
base_for_goc = {
"assets": ir_with_attestations.get("assets", []),
"objectives": ir_with_attestations.get("objectives", []),
"constraints": ir_with_attestations.get("constraints", {}),
"execution_policy": ir_with_attestations.get("execution_policy", []),
}
digest = hashlib.sha256(json.dumps(base_for_goct, sort_keys=True).encode("utf-8")).hexdigest()
digest = hashlib.sha256(json.dumps(base_for_goc, sort_keys=True).encode("utf-8")).hexdigest()
except Exception:
digest = base_digest
# Prefer a registry-driven skeleton when available
if build_minimal_goC_skeleton is not None:
try:
return build_minimal_goC_skeleton(digest, base_timestamp)
except Exception:
pass
# Fallback to a lightweight, self-contained skeleton
# Include both a source_digest and a registry_digest to enable auditability
# and to satisfy downstream tests that expect a registry_digest field.
return {
"version": "0.1",
"contracts": [],
"metadata": {
"generated_at": base_timestamp,
"registry_digest": digest,
"source_digest": digest,
skeleton = build_minimal_goC_skeleton(digest, base_timestamp)
skeleton["contracts"] = [
{
"name": "equiir-core",
"version": str(ir_with_attestations.get("version", DEFAULT_VERSION)),
"adapter": "compiler",
},
}
{
"name": "reference-backtest",
"version": "0.1",
"adapter": "python_backtester",
},
]
skeleton["metadata"]["source_digest"] = digest
return skeleton
__all__ = ["parse_dsl_to_ir", "to_json"]
__all__ = ["compile_strategy", "parse_dsl_to_ir", "to_json"]

17
equicompiler_algebraic_portfolio_dsl_to_/delta_sync.py
View File

@ -13,7 +13,8 @@ from typing import Dict, Any
def _digest(ir: Dict[str, Any]) -> str:
"""Compute a stable digest for a given IR dictionary."""
return hashlib.sha256(json.dumps(ir, sort_keys=True).encode("utf-8")).hexdigest()
canonical = {k: v for k, v in ir.items() if k not in {"attestations", "graph_of_contracts", "delta_sync"}}
return hashlib.sha256(json.dumps(canonical, sort_keys=True).encode("utf-8")).hexdigest()
def delta_sync(old_ir: Dict[str, Any], new_ir: Dict[str, Any]) -> Dict[str, Any]:
@ -30,9 +31,13 @@ def delta_sync(old_ir: Dict[str, Any], new_ir: Dict[str, Any]) -> Dict[str, Any]
"assets_added": sorted(list(new_assets - old_assets)),
"assets_removed": sorted(list(old_assets - new_assets)),
"constraints_changed": {}, # naive: compute by key presence and value change
"risk_budget_changed": {},
"market_microstructure_changed": {},
"regulatory_changed": {},
"execution_policy_added": [],
"execution_policy_removed": [],
"objectives_changed": {}, # {<objective_key>: {"old": <old>, "new": <new>}}
"plan_changed": old_ir.get("plan") != new_ir.get("plan"),
"digest": _digest(new_ir),
}
@ -45,6 +50,16 @@ def delta_sync(old_ir: Dict[str, Any], new_ir: Dict[str, Any]) -> Dict[str, Any]
if ov != nv:
delta["constraints_changed"][k] = {"old": ov, "new": nv}
for field_name in ("risk_budget", "market_microstructure", "regulatory"):
old_field = old_ir.get(field_name, {}) or {}
new_field = new_ir.get(field_name, {}) or {}
all_field_keys = set(old_field.keys()) | set(new_field.keys())
for k in all_field_keys:
ov = old_field.get(k)
nv = new_field.get(k)
if ov != nv:
delta[f"{field_name}_changed"][k] = {"old": ov, "new": nv}
# Execution policy delta (order-insensitive for MVP)
old_policy = set(old_ir.get("execution_policy", []) or [])
new_policy = set(new_ir.get("execution_policy", []) or [])

12
pyproject.toml
View File

@ -8,8 +8,18 @@ version = "0.1.0"
description = "Algebraic Portfolio DSL to verifiable low-latency market strategy compiler (MVP)"
readme = "README.md"
requires-python = ">=3.9"
license = {text = "MIT"}
license = "MIT"
authors = [ { name = "OpenCode Team" } ]
classifiers = [
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3 :: Only",
"Operating System :: OS Independent",
"Topic :: Office/Business :: Financial",
]
[project.urls]
Homepage = "https://example.com/equicompiler"
Repository = "https://example.com/equicompiler.git"
[tool.setuptools.packages.find]
where = ["."]

1
test.sh
View File

@ -17,6 +17,7 @@ else
fi
echo "Building package..."
python3 -m pip install build
python3 -m build
echo "All tests executed."

60
tests/test_compiler_pipeline.py Normal file
View File

@ -0,0 +1,60 @@
import unittest
from equicompiler_algebraic_portfolio_dsl_to_.backends.python_backtester import run_backtest
from equicompiler_algebraic_portfolio_dsl_to_.core import parse_dsl_to_ir
from equicompiler_algebraic_portfolio_dsl_to_.delta_sync import delta_sync
class TestCompilerPipeline(unittest.TestCase):
def test_richer_dsl_sections_are_parsed(self):
dsl = (
"version: 0.3\n"
"assets: AAPL@0.6, MSFT@0.4\n"
"objectives: maximize_return; minimize_risk\n"
"constraints: max_drawdown=0.15, var=0.95, cvar=0.9\n"
"risk_budget: max_var=0.1, max_exposure=0.8\n"
"execution_policy: offline, delta_sync, rebalance=1d\n"
"regulatory: venue=us, suitability=true"
)
ir = parse_dsl_to_ir(dsl)
self.assertEqual(ir["version"], "0.3")
self.assertEqual(ir["assets"], ["AAPL", "MSFT"])
self.assertEqual(ir["asset_weights"]["AAPL"], 0.6)
self.assertEqual(ir["risk_budget"]["max_var"], 0.1)
self.assertTrue(ir["regulatory"]["suitability"])
self.assertIn("plan", ir)
self.assertGreaterEqual(len(ir["plan"]["phases"]), 3)
self.assertIn("lineage", ir)
def test_reference_backtest_produces_metrics(self):
ir = parse_dsl_to_ir(
"assets: AAPL, MSFT\n"
"objectives: maximize_return\n"
"constraints: max_drawdown=0.2"
)
result = run_backtest(ir)
self.assertIn("metrics", result)
self.assertIn("total_return", result["metrics"])
self.assertIn("max_drawdown", result["metrics"])
self.assertIsInstance(result["objective_met"], bool)
def test_delta_sync_detects_plan_changes(self):
base = parse_dsl_to_ir(
"assets: AAPL, MSFT\n"
"objectives: maximize_return\n"
"constraints: max_drawdown=0.2"
)
updated = parse_dsl_to_ir(
"assets: AAPL, MSFT, GOOG\n"
"objectives: maximize_return\n"
"constraints: max_drawdown=0.15"
)
delta = delta_sync(base, updated)
self.assertIn("assets_added", delta)
self.assertIn("GOOG", delta["assets_added"])
self.assertTrue(delta["plan_changed"])
self.assertIn("constraints_changed", delta)
if __name__ == "__main__":
unittest.main()