| |
| """ |
| AirMicroDrip Audit Integration |
| Connects the audit framework to the AirMicroDrip perpetual futures system. |
| Provides continuous monitoring, health checks, and compliance verification. |
| """ |
|
|
| import os |
| import sys |
| from datetime import datetime |
| from typing import Dict, Any, Optional |
|
|
| |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) |
|
|
| from audit_framework import ( |
| AuditFramework, |
| AuditCategory, |
| AuditSeverity, |
| AuditStatus, |
| ) |
|
|
|
|
| class AirMicroDripAuditor: |
| """Audit wrapper for AirMicroDrip systems.""" |
|
|
| def __init__(self, db_path: str = "airmicrodrip_audit.db"): |
| self.audit = AuditFramework() |
| self.db_path = db_path |
| self._register_airmicrodrip_checks() |
|
|
| def _register_airmicrodrip_checks(self): |
| """Register AirMicroDrip-specific audit checks.""" |
| from audit_framework import AuditCheck |
|
|
| extra_checks = [ |
| AuditCheck( |
| check_id="amd_001", |
| name="Liquidity Provider Health", |
| description="Verify at least one active LLM inference provider", |
| category=AuditCategory.AVAILABILITY, |
| severity=AuditSeverity.HIGH, |
| ), |
| AuditCheck( |
| check_id="amd_002", |
| name="Synthetic Liquidity Depth", |
| description="Verify total synthetic liquidity exceeds minimum threshold", |
| category=AuditCategory.ACCURACY, |
| severity=AuditSeverity.HIGH, |
| ), |
| AuditCheck( |
| check_id="amd_003", |
| name="Perpetual Engine Consistency", |
| description="Verify mark prices and index prices are within tolerance", |
| category=AuditCategory.ACCURACY, |
| severity=AuditSeverity.CRITICAL, |
| ), |
| AuditCheck( |
| check_id="amd_004", |
| name="Funding Rate Bounds", |
| description="Verify funding rates are within configured min/max", |
| category=AuditCategory.ACCURACY, |
| severity=AuditSeverity.MEDIUM, |
| ), |
| AuditCheck( |
| check_id="amd_005", |
| name="Liquidation Backlog", |
| description="Verify no positions are stuck in liquidation queue", |
| category=AuditCategory.INTEGRITY, |
| severity=AuditSeverity.CRITICAL, |
| ), |
| AuditCheck( |
| check_id="amd_006", |
| name="Order Book Spread", |
| description="Verify bid-ask spread is within acceptable range", |
| category=AuditCategory.PERFORMANCE, |
| severity=AuditSeverity.MEDIUM, |
| ), |
| ] |
|
|
| for check in extra_checks: |
| self.audit.checks[check.check_id] = check |
|
|
| def check_liquidity_providers(self, registry) -> Dict[str, Any]: |
| """Run liquidity provider health check.""" |
| providers = registry.get_all_providers(status="active") |
| if not providers: |
| return { |
| "status": AuditStatus.FAILED, |
| "message": "No active liquidity providers", |
| "details": {"active_count": 0}, |
| } |
| return { |
| "status": AuditStatus.PASSED, |
| "message": f"{len(providers)} active liquidity providers", |
| "details": {"active_count": len(providers)}, |
| } |
|
|
| def check_liquidity_depth(self, converter) -> Dict[str, Any]: |
| """Run synthetic liquidity depth check.""" |
| total = converter.get_total_liquidity() |
| total_usd = total.get("total_usd", 0.0) |
| min_liquidity = float(os.environ.get("MIN_LIQUIDITY_USD", 10000.0)) |
|
|
| if total_usd < min_liquidity: |
| return { |
| "status": AuditStatus.FAILED, |
| "message": f"Total liquidity ${total_usd:.2f} below minimum ${min_liquidity:.2f}", |
| "details": {"total_usd": total_usd, "minimum": min_liquidity}, |
| } |
| return { |
| "status": AuditStatus.PASSED, |
| "message": f"Total liquidity ${total_usd:.2f} above minimum", |
| "details": {"total_usd": total_usd, "by_market": total.get("by_market", {})}, |
| } |
|
|
| def check_mark_price_consistency(self, trading_engine, tolerance: float = 0.02) -> Dict[str, Any]: |
| """Verify mark prices are close to index prices.""" |
| inconsistent = [] |
| for market, state in trading_engine.market_states.items(): |
| if state.index_price == 0: |
| continue |
| deviation = abs(state.mark_price - state.index_price) / state.index_price |
| if deviation > tolerance: |
| inconsistent.append({ |
| "market": market, |
| "mark": state.mark_price, |
| "index": state.index_price, |
| "deviation": deviation, |
| }) |
|
|
| if inconsistent: |
| return { |
| "status": AuditStatus.FAILED, |
| "message": f"{len(inconsistent)} market(s) with price deviation > {tolerance:.1%}", |
| "details": {"inconsistent": inconsistent}, |
| } |
| return { |
| "status": AuditStatus.PASSED, |
| "message": "Mark prices consistent with index prices", |
| "details": {"markets_checked": len(trading_engine.market_states)}, |
| } |
|
|
| def check_funding_rate_bounds(self, funding_engine) -> Dict[str, Any]: |
| """Verify funding rates within bounds.""" |
| from funding_rate_engine import FUNDING_CONFIG |
|
|
| out_of_bounds = [] |
| for market in funding_engine.trading_engine.market_states: |
| rate = funding_engine.calculate_funding_rate(market) |
| if rate < FUNDING_CONFIG["min_funding_rate"] or rate > FUNDING_CONFIG["max_funding_rate"]: |
| out_of_bounds.append({"market": market, "rate": rate}) |
|
|
| if out_of_bounds: |
| return { |
| "status": AuditStatus.FAILED, |
| "message": f"{len(out_of_bounds)} funding rate(s) out of bounds", |
| "details": {"out_of_bounds": out_of_bounds}, |
| } |
| return { |
| "status": AuditStatus.PASSED, |
| "message": "All funding rates within bounds", |
| "details": {"markets_checked": len(funding_engine.trading_engine.market_states)}, |
| } |
|
|
| def check_liquidation_backlog(self, liq_system) -> Dict[str, Any]: |
| """Check for stuck liquidations.""" |
| at_risk = liq_system.get_at_risk_positions() |
| if len(at_risk) > 10: |
| return { |
| "status": AuditStatus.WARNING, |
| "message": f"{len(at_risk)} positions at risk — possible backlog", |
| "details": {"at_risk_count": len(at_risk)}, |
| } |
| return { |
| "status": AuditStatus.PASSED, |
| "message": f"Liquidation queue healthy ({len(at_risk)} at risk)", |
| "details": {"at_risk_count": len(at_risk)}, |
| } |
|
|
| def check_orderbook_spread(self, trading_engine, max_spread_bps: float = 50.0) -> Dict[str, Any]: |
| """Verify bid-ask spreads are within tolerance.""" |
| wide_spreads = [] |
| for market, ob in trading_engine.order_books.items(): |
| best_bid = ob.get_best_bid() |
| best_ask = ob.get_best_ask() |
| if best_bid and best_ask and best_bid > 0: |
| spread_bps = ((best_ask - best_bid) / best_bid) * 10000 |
| if spread_bps > max_spread_bps: |
| wide_spreads.append({"market": market, "spread_bps": spread_bps}) |
|
|
| if wide_spreads: |
| return { |
| "status": AuditStatus.WARNING, |
| "message": f"{len(wide_spreads)} market(s) with wide spread", |
| "details": {"wide_spreads": wide_spreads}, |
| } |
| return { |
| "status": AuditStatus.PASSED, |
| "message": "Order book spreads within tolerance", |
| "details": {"markets_checked": len(trading_engine.order_books)}, |
| } |
|
|
| def run_airmicrodrip_audit( |
| self, |
| registry=None, |
| converter=None, |
| trading_engine=None, |
| funding_engine=None, |
| liq_system=None, |
| ) -> Dict[str, Any]: |
| """Run the full AirMicroDrip audit suite.""" |
| ctx: Dict[str, Any] = {} |
|
|
| if registry: |
| ctx["liquidity_providers"] = self.check_liquidity_providers(registry) |
| if converter: |
| ctx["liquidity_depth"] = self.check_liquidity_depth(converter) |
| if trading_engine: |
| ctx["price_consistency"] = self.check_mark_price_consistency(trading_engine) |
| if funding_engine: |
| ctx["funding_bounds"] = self.check_funding_rate_bounds(funding_engine) |
| if liq_system: |
| ctx["liquidation_backlog"] = self.check_liquidation_backlog(liq_system) |
| if trading_engine: |
| ctx["orderbook_spread"] = self.check_orderbook_spread(trading_engine) |
|
|
| |
| for check_name, result in ctx.items(): |
| status = result.get("status", AuditStatus.SKIPPED) |
| self.audit.log( |
| category=AuditCategory.INTEGRITY, |
| severity=AuditSeverity.HIGH if status == AuditStatus.FAILED else AuditSeverity.INFO, |
| status=status, |
| message=result.get("message", f"{check_name} check completed"), |
| details=result.get("details", {}), |
| actor="airmicrodrip_auditor", |
| component=check_name, |
| ) |
|
|
| |
| base_report = self.audit.run_audit(context=ctx) |
|
|
| return { |
| "base_report_id": base_report.report_id, |
| "overall_score": base_report.overall_score, |
| "airmicrodrip_checks": ctx, |
| "system_health": self.audit.get_system_health(), |
| } |
|
|
|
|
| if __name__ == "__main__": |
| |
| auditor = AirMicroDripAuditor() |
| print("AirMicroDrip Auditor initialized with checks:") |
| for cid, check in auditor.audit.checks.items(): |
| print(f" {cid}: {check.name} ({check.category.value}, {check.severity.value})") |
| print(f"\nTotal checks registered: {len(auditor.audit.checks)}") |
|
|
