CI: Added tests for ml orchestration scripts

Author: AlmostAnna

.github/workflows/test.yml

@@ -42,9 +42,5 @@ jobs:
 
     - name: Run tests
       run: |
-        pytest tests/ -v --cov=src --cov-report=xml --cov-report=term
+        pytest tests/ -v --cov=src --cov=ml --cov-report=xml --cov-report=term
       shell: bash -l {0}
-
-
-
-

pyproject.toml

@@ -45,6 +45,7 @@ addopts = """
     --tb=short
     --doctest-modules
     --cov=src/quantlab
+    --cov=ml
     --cov-report=term-missing
     --cov-report=html
 """
@@ -55,6 +56,11 @@ markers = [
     "slow: Tests that take >1s (skip by default)",
 ]
 
+[tool.coverage.run]
+source = ["src", "ml"]
+omit = ["tests/*", "*/__pycache__/*", "*.pyc"]
+disable_warnings = ["no-data-collected", "module-not-imported"]
+
 [tool.black]
 line-length = 88
 target-version = ['py38']

tests/ml/test_metrics/test_pnl.py

@@ -0,0 +1,112 @@
+"""Tests for P&L computation functions."""
+
+import pytest
+import torch
+
+from quantlab.ml.metrics.pnl import compute_pnl_with_tx, decompose_pnl
+
+
+def test_compute_pnl_with_tx_basic():
+    """Test basic P&L calculation."""
+    S = torch.tensor([[100.0, 105.0, 110.0]])  # Single path, 3 time steps
+    K = 105.0
+    phi = torch.tensor([[0.5, 0.6]])  # Holdings at t0, t1
+    lambda_tx = 0.01
+
+    pnl = compute_pnl_with_tx(S, K, phi, lambda_tx)
+
+    expected_payoff = max(110.0 - 105.0, 0)  # 5.0
+    expected_hedging_gain = 0.5 * (105.0 - 100.0) + 0.6 * (
+        110.0 - 105.0
+    )  # 2.5 + 3.0 = 5.5
+    expected_tx_cost = 0.01 * abs(0.6 - 0.5) * 105.0  # 0.01 * 0.1 * 105 = 0.105
+    expected_pnl = expected_payoff - expected_hedging_gain - expected_tx_cost
+
+    assert torch.isclose(pnl[0], torch.tensor(expected_pnl), atol=1e-5)
+
+
+def test_compute_pnl_with_tx_no_transactions():
+    """Test P&L with no trading."""
+    S = torch.tensor([[100.0, 105.0, 110.0]])
+    K = 105.0
+    phi = torch.tensor([[0.5, 0.5]])  # No change in holdings
+    lambda_tx = 0.01
+
+    pnl = compute_pnl_with_tx(S, K, phi, lambda_tx)
+
+    # Should have no transaction costs
+    expected_payoff = 5.0
+    expected_hedging_gain = 0.5 * (105.0 - 100.0) + 0.5 * (110.0 - 105.0)  # 5.0
+    expected_pnl = expected_payoff - expected_hedging_gain  # 0.0
+
+    assert torch.isclose(pnl[0], torch.tensor(expected_pnl), atol=1e-5)
+
+
+def test_decompose_pnl():
+    """Test P&L decomposition."""
+    S = torch.tensor([[100.0, 105.0, 110.0]])
+    K = 105.0
+    phi = torch.tensor([[0.5, 0.6]])
+    lambda_tx = 0.01
+
+    total_pnl, hedging_gain, tx_cost = decompose_pnl(S, K, phi, lambda_tx)
+
+    assert total_pnl.shape == (1,)
+    assert hedging_gain.shape == (1,)
+    assert tx_cost.shape == (1,)
+
+    # Verify decomposition
+    pnl_direct = compute_pnl_with_tx(S, K, phi, lambda_tx)
+    assert torch.allclose(total_pnl, pnl_direct)
+
+
+def test_pnl_edge_cases():
+    """Test edge cases."""
+    # Out-of-the-money option
+    S = torch.tensor([[100.0, 95.0, 90.0]])
+    K = 105.0
+    phi = torch.tensor([[0.0, 0.0]])
+    lambda_tx = 0.0
+
+    pnl = compute_pnl_with_tx(S, K, phi, lambda_tx)
+    assert torch.isclose(pnl[0], torch.tensor(0.0))  # Zero payoff, zero hedging gain
+
+    # At-the-money
+    S = torch.tensor([[100.0, 100.0, 100.0]])
+    K = 100.0
+    pnl = compute_pnl_with_tx(S, K, phi, lambda_tx)
+    assert torch.isclose(pnl[0], torch.tensor(0.0))
+
+
+@pytest.mark.parametrize(
+    "shape",
+    [
+        ((1, 2), (1, 1)),  # Minimal case
+        ((10, 5), (10, 4)),  # Multiple paths
+        ((1, 100), (1, 99)),  # Long time series
+    ],
+)
+def test_pnl_shapes(shape):
+    """Test various input shapes."""
+    S_shape, phi_shape = shape
+    S = torch.randn(S_shape)
+    K = 100.0
+    phi = torch.rand(phi_shape)
+    lambda_tx = 0.01
+
+    pnl = compute_pnl_with_tx(S, K, phi, lambda_tx)
+    assert pnl.shape[0] == S_shape[0]  # Same number of samples
+
+
+def test_pnl_gradient_flow():
+    """Test that gradients flow through P&L computation."""
+    S = torch.tensor([[100.0, 105.0, 110.0]], requires_grad=True)
+    K = torch.tensor(105.0, requires_grad=True)
+    phi = torch.tensor([[0.5, 0.6]], requires_grad=False)
+    lambda_tx = 0.01
+
+    pnl = compute_pnl_with_tx(S, K, phi, lambda_tx)
+    pnl.sum().backward()
+
+    assert S.grad is not None
+    assert K.grad is not None

tests/ml/test_train.py

@@ -0,0 +1,113 @@
+"""Tests for training pipeline - functionality verification."""
+
+import subprocess
+import sys
+from pathlib import Path
+
+import torch
+
+
+def test_train_cli_help():
+    """Test that the training CLI can be invoked."""
+    project_root = Path(__file__).parent.parent.parent
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            """
+import sys
+sys.path.insert(0, '.')
+import ml.train
+import argparse
+parser = argparse.ArgumentParser(description='Deep Hedging')
+parser.add_argument('--data_type_model', type=str, default='gbm')
+print('Parser created successfully')
+        """,
+        ],
+        capture_output=True,
+        text=True,
+        cwd=project_root,
+    )
+
+    assert result.returncode == 0, f"Failed to import train module: {result.stderr}"
+
+
+def test_prepare_inputs_logic():
+    """Test the prepare_inputs logic directly in the test."""
+
+    def prepare_inputs(S, K, T, M, device="cpu"):
+        """Prepare input batches for training."""
+        t_grid = torch.linspace(0, T - T / M, M, device=device)
+        tau_grid = T - t_grid
+        N = S.size(0)
+        tau_batch = tau_grid.unsqueeze(0).expand(N, -1)
+        moneyness_batch = S[:, :-1] / K
+        return tau_batch.reshape(-1), moneyness_batch.reshape(-1)
+
+    S = torch.tensor([[100.0, 105.0, 110.0]])  # Shape: (1, 3) - N=1, M=2
+    K = 105.0  # Scalar
+    T = 1.0
+    M = 2  # 2 time steps between 0 and T
+
+    tau_flat, moneyness_flat = prepare_inputs(S, K, T, M)
+    assert tau_flat.shape == torch.Size(
+        [2]
+    ), f"Expected shape [2], got {tau_flat.shape}"
+    assert moneyness_flat.shape == torch.Size(
+        [2]
+    ), f"Expected shape [2], got {moneyness_flat.shape}"
+
+    # Check individual values
+    assert torch.allclose(
+        tau_flat[0], torch.tensor(1.0), atol=1e-6
+    ), f"First tau value incorrect: {tau_flat[0]}"
+    assert torch.allclose(
+        tau_flat[1], torch.tensor(0.5), atol=1e-6
+    ), f"Second tau value incorrect: {tau_flat[1]}"
+    assert torch.allclose(
+        moneyness_flat[0], torch.tensor(100.0 / 105.0), atol=1e-6
+    ), f"First moneyness value incorrect: {moneyness_flat[0]}"
+    assert torch.allclose(
+        moneyness_flat[1], torch.tensor(105.0 / 105.0), atol=1e-6
+    ), f"Second moneyness value incorrect: {moneyness_flat[1]}"
+
+
+def test_train_script_syntax():
+    """Test that the train script has valid Python syntax."""
+    project_root = Path(__file__).parent.parent.parent
+    train_file = project_root / "ml" / "train.py"
+
+    with open(train_file, "r") as f:
+        code = f.read()
+
+    # This will raise SyntaxError if there are syntax issues
+    compile(code, str(train_file), "exec")
+
+
+def test_train_can_run_minimal():
+    """Test that train script can run with minimal configuration."""
+    project_root = Path(__file__).parent.parent.parent
+
+    # Create a minimal test that doesn't actually train but validates imports
+    test_code = """
+import sys
+sys.path.insert(0, '.')
+sys.path.insert(0, './src')
+
+# Test imports work
+from quantlab.ml.models.hedge_net import HedgeNet
+from quantlab.ml.metrics.pnl import compute_pnl_with_tx
+
+# Verify basic functionality
+net = HedgeNet(hidden_dim=16)
+print('Basic imports and instantiation successful')
+"""
+
+    result = subprocess.run(
+        [sys.executable, "-c", test_code],
+        capture_output=True,
+        text=True,
+        cwd=project_root,
+    )
+
+    assert result.returncode == 0, f"Basic functionality test failed: {result.stderr}"