76 lines
3.1 KiB
Python
76 lines
3.1 KiB
Python
import unittest
|
|
from typing import cast
|
|
|
|
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 = cast(dict, 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)
|
|
self.assertIn("verification", ir)
|
|
self.assertEqual(ir["verification"]["status"], "verified")
|
|
self.assertIn("prove_constraints", ir["plan"]["phases"][1]["actions"])
|
|
|
|
def test_verification_flags_invalid_weights(self):
|
|
ir = cast(dict, parse_dsl_to_ir(
|
|
"assets: AAPL@0.9, MSFT@0.2\n"
|
|
"objectives: maximize_return\n"
|
|
"constraints: max_drawdown=0.2"
|
|
))
|
|
self.assertEqual(ir["verification"]["status"], "needs_review")
|
|
self.assertIn("asset_weights_do_not_sum_to_one", ir["verification"]["issues"])
|
|
self.assertGreaterEqual(len(ir["constraint_spec"]), 1)
|
|
|
|
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 = cast(dict, parse_dsl_to_ir(
|
|
"assets: AAPL, MSFT\n"
|
|
"objectives: maximize_return\n"
|
|
"constraints: max_drawdown=0.2"
|
|
))
|
|
updated = cast(dict, 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)
|
|
self.assertTrue(delta["verification_changed"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|