ci.yml

name: CI

on:
  push:
    branches: ["main"]
  pull_request:
    branches: ["**"]

env:
  PYTHON_VERSION: "3.12"

jobs:
  lint:
    runs-on: ubuntu-22.04

    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Install dependencies (pinned by tests/constraints.txt)
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt -c tests/constraints.txt

      - name: Run formatter
        run: black --check .

      # - name: Run linter
      #   run: flake8 .
      
  test:
    runs-on: ubuntu-22.04
    needs: lint   # only run tests if lint passes

    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Install dependencies (pinned by tests/constraints.txt)
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt -c tests/constraints.txt

      - name: Check SVI determinism (hard gate)
        env:
          OMP_NUM_THREADS: "1"
          OPENBLAS_NUM_THREADS: "1"
          MKL_NUM_THREADS: "1"
          NUMEXPR_NUM_THREADS: "1"
        run: |
          PYTHONPATH=. python - <<'PY'
          import numpy as np
          import pandas as pd
          import scipy
          from oipd import MarketInputs, VolCurve

          print("numpy", np.__version__)
          print("scipy", scipy.__version__)
          print("pandas", pd.__version__)

          df = pd.read_csv("tests/data/AAPL_data.csv")
          df = df[df["expiration"] == "2026-01-16"]

          market = MarketInputs(
              valuation_date=pd.Timestamp("2025-01-15"),
              risk_free_rate=0.045,
              underlying_price=220.0,
          )

          column_mapping = {
              "strike": "strike",
              "last_price": "last_price",
              "type": "option_type",
              "bid": "bid",
              "ask": "ask",
              "expiration": "expiry",
          }

          param_fields = (
              "a",
              "b",
              "rho",
              "m",
              "sigma",
              "forward",
              "maturity_years",
          )
          diagnostic_fields = (
              "objective",
              "rmse_weighted",
              "rmse_unweighted",
              "envelope_violations_pct",
              "chosen_start_origin",
              "chosen_start_index",
          )

          def is_nan(value):
              try:
                  return bool(np.isnan(value))
              except TypeError:
                  return False

          def values_equal(left, right):
              if isinstance(left, np.ndarray) or isinstance(right, np.ndarray):
                  left_array = np.asarray(left, dtype=object)
                  right_array = np.asarray(right, dtype=object)
                  if left_array.shape != right_array.shape:
                      return False
                  return all(
                      values_equal(left_item, right_item)
                      for left_item, right_item in zip(
                          left_array.flat, right_array.flat
                      )
                  )
              if isinstance(left, (list, tuple)) and isinstance(right, (list, tuple)):
                  if len(left) != len(right):
                      return False
                  return all(
                      values_equal(left_item, right_item)
                      for left_item, right_item in zip(left, right)
                  )
              if isinstance(left, dict) and isinstance(right, dict):
                  if left.keys() != right.keys():
                      return False
                  return all(
                      values_equal(left[key], right[key]) for key in left.keys()
                  )
              if is_nan(left) and is_nan(right):
                  return True
              return left == right

          def run_fit():
              vc = VolCurve(method="svi")
              vc.method_options = {"random_seed": 42}
              vc.fit(df, market, column_mapping=column_mapping)

              payload = {field: vc.params.get(field) for field in param_fields}
              diagnostics = getattr(vc, "diagnostics", None)
              for field in diagnostic_fields:
                  payload[field] = getattr(diagnostics, field, None)
              return payload

          payload_1 = run_fit()
          payload_2 = run_fit()

          print("SVI payload run1", payload_1)
          print("SVI payload run2", payload_2)

          mismatches = []
          for field in payload_1.keys():
              run1_value = payload_1[field]
              run2_value = payload_2[field]
              if not values_equal(run1_value, run2_value):
                  mismatches.append(
                      {
                          "field": field,
                          "run1": repr(run1_value),
                          "run2": repr(run2_value),
                      }
                  )

          if mismatches:
              print("SVI determinism check FAILED")
              for mismatch in mismatches:
                  print(
                      "field={field} run1={run1} run2={run2}".format(**mismatch)
                  )
              raise SystemExit(1)

          print("SVI determinism check passed: exact match across 2 seeded runs")
          PY

      - name: Run tests (parallel, excluding regression)
        env:
          OMP_NUM_THREADS: "1"
          OPENBLAS_NUM_THREADS: "1"
          MKL_NUM_THREADS: "1"
          NUMEXPR_NUM_THREADS: "1"
        run: pytest -q -n auto --dist=loadscope tests --ignore=tests/regression

      - name: Run regression golden master (serial)
        run: pytest -q tests/regression/test_golden_master.py