kcolbchain lab. zero-knowledge ML inference

import torch
import torch.nn as nn
import numpy as np

# Simple 2-layer network for liquidation risk prediction
class LiquidationRiskModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(4, 16),
            nn.ReLU(),
            nn.Linear(16, 8),
            nn.ReLU(),
            nn.Linear(8, 1),
            nn.Sigmoid(),
        )

    def forward(self, x):
        return self.net(x)

# Generate synthetic training data
np.random.seed(42)
N = 2000
collateral_ratio = np.random.uniform(1.0, 3.0, N)
borrow_apy = np.random.uniform(0.01, 0.30, N)
volatility = np.random.uniform(0.05, 0.80, N)
utilization = np.random.uniform(0.3, 0.99, N)

# Label: high risk if collateral ratio is low AND volatility is high
risk = ((collateral_ratio < 1.5) & (volatility > 0.4)).astype(np.float32)

X = np.column_stack([collateral_ratio, borrow_apy, volatility, utilization]).astype(np.float32)
y = risk.reshape(-1, 1)

X_t, y_t = torch.tensor(X), torch.tensor(y)

# Train
model = LiquidationRiskModel()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
loss_fn = nn.BCELoss()

for epoch in range(100):
    pred = model(X_t)
    loss = loss_fn(pred, y_t)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()
    if epoch % 25 == 0:
        print(f"Epoch {epoch}, loss: {loss.item():.4f}")

# Export to ONNX (required by EZKL)
dummy_input = torch.randn(1, 4)
torch.onnx.export(
    model, dummy_input, "liquidation_risk.onnx",
    input_names=["features"],
    output_names=["risk_score"],
    dynamic_axes={"features": {0: "batch"}},
)
print("Exported to liquidation_risk.onnx")

# Save a test input for proving
test_input = torch.tensor([[1.2, 0.15, 0.65, 0.85]])  # high risk case
with torch.no_grad():
    test_output = model(test_input)
print(f"Test prediction: {test_output.item():.4f} (should be close to 1.0 = high risk)")

# Save test input as JSON for EZKL
import json
json.dump({"input_data": [test_input.tolist()]}, open("input.json", "w"))

# Step 1: Generate settings (calibrate the circuit)
ezkl gen-settings \
  --model liquidation_risk.onnx \
  --settings settings.json

# Step 2: Calibrate with sample data for optimal quantization
ezkl calibrate-settings \
  --model liquidation_risk.onnx \
  --settings settings.json \
  --data input.json

# Step 3: Compile the model to a circuit
ezkl compile-circuit \
  --model liquidation_risk.onnx \
  --compiled-circuit model.compiled \
  --settings settings.json

# Step 4: Run the trusted setup (generates proving + verification keys)
ezkl setup \
  --compiled-circuit model.compiled \
  --pk-path pk.key \
  --vk-path vk.key

# Step 5: Generate the proof
ezkl prove \
  --compiled-circuit model.compiled \
  --pk-path pk.key \
  --witness input.json \
  --proof-path proof.json

# Step 6: Verify locally (sanity check before on-chain)
ezkl verify \
  --proof-path proof.json \
  --vk-path vk.key \
  --settings settings.json

echo "Proof verified locally!"

# Generate the Solidity verifier contract
ezkl create-evm-verifier \
  --vk-path vk.key \
  --settings settings.json \
  --sol-code-path Verifier.sol

# The generated Verifier.sol contains the verification key
# baked into the contract -- no external dependencies.

# Deploy with Foundry
forge create Verifier.sol:Halo2Verifier \
  --rpc-url https://sepolia.base.org \
  --private-key $PRIVATE_KEY

# => Deployer: 0x...your_address...
# => Deployed to: 0x...verifier_address...

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.20;

interface IVerifier {
    function verify(
        bytes calldata proof,
        uint256[] calldata instances
    ) external view returns (bool);
}

contract LiquidationOracle {
    IVerifier public immutable verifier;

    // Stores the latest verified risk prediction
    struct RiskPrediction {
        uint256 riskScore;     // fixed-point, scaled by 1e18
        uint256 timestamp;
        address submitter;
    }

    mapping(address => RiskPrediction) public predictions;

    event RiskUpdated(address indexed position, uint256 riskScore, address submitter);

    constructor(address _verifier) {
        verifier = IVerifier(_verifier);
    }

    /// @notice Submit a verified ML risk prediction
    /// @param position The lending position address
    /// @param proof The EZKL proof bytes
    /// @param instances Public inputs/outputs of the circuit
    function submitPrediction(
        address position,
        bytes calldata proof,
        uint256[] calldata instances
    ) external {
        // Verify the ZK proof on-chain
        require(verifier.verify(proof, instances), "Invalid ZKML proof");

        // instances[0..3] = input features (collateral ratio, etc.)
        // instances[4] = model output (risk score)
        uint256 riskScore = instances[4];

        predictions[position] = RiskPrediction({
            riskScore: riskScore,
            timestamp: block.timestamp,
            submitter: msg.sender
        });

        emit RiskUpdated(position, riskScore, msg.sender);
    }

    /// @notice Check if a position is high risk (score > 0.7)
    function isHighRisk(address position) external view returns (bool) {
        RiskPrediction memory pred = predictions[position];
        require(pred.timestamp > 0, "No prediction");
        require(block.timestamp - pred.timestamp < 1 hours, "Prediction stale");
        return pred.riskScore > 7e17; // 0.7 scaled by 1e18
    }
}

from giza_agents import GizaAgent, AgentConfig
from giza_agents.model import GizaModel

# Step 1: Transpile ONNX model to Cairo via Orion
config = AgentConfig(
    model_path="liquidation_risk.onnx",
    framework="CAIRO",
)

model = GizaModel(config)
model.transpile()  # converts ONNX -> Cairo/Orion representation

# Step 2: Create a verifiable agent
agent = GizaAgent(
    model=model,
    chain="starknet-sepolia",  # STARK verification is native on Starknet
)

# Step 3: Run inference with proof generation
input_data = [1.2, 0.15, 0.65, 0.85]
result = agent.predict(input_data, verify=True)

print(f"Prediction: {result.output}")
print(f"Proof ID: {result.proof_id}")
print(f"Verifiable on-chain: {result.verification_url}")

# Step 4: Verify on Starknet
verification = agent.verify_on_chain(result.proof_id)
print(f"On-chain verification tx: {verification.tx_hash}")