6c7eed7580
Fresh history cut from cynarAI/evoo antigravity-handoff/project-timeline (Voice First + project timeline). Brand docs/titles as evOo; keep @rakazo/* packages as tech debt. Excludes node_modules, caches, secrets, voice blobs.
63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
"""
|
|
Apex Trading Engine Core
|
|
Event loop that receives signals, checks risk guardrails, and executes orders.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Dict, Any
|
|
|
|
from pathlib import Path
|
|
|
|
SRC_DIR = Path(__file__).resolve().parent.parent
|
|
if str(SRC_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SRC_DIR))
|
|
|
|
try:
|
|
from risk.risk_guard import ApexRiskGuard
|
|
from strategies.momentum import MomentumStrategy
|
|
except ImportError:
|
|
from ..risk.risk_guard import ApexRiskGuard
|
|
from ..strategies.momentum import MomentumStrategy
|
|
|
|
class ApexEngine:
|
|
def __init__(self, config_path: str):
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|
self.config = json.load(f)
|
|
|
|
self.risk_guard = ApexRiskGuard(self.config)
|
|
self.momentum_strategy = MomentumStrategy(self.config.get("strategies", {}).get("momentum"))
|
|
self.open_positions = []
|
|
self.current_drawdown_pct = 0.0
|
|
|
|
def process_tick(self, symbol: str, prices: list, volume: list) -> Dict[str, Any]:
|
|
"""Processes market data tick through strategy and risk guard."""
|
|
signal = self.momentum_strategy.generate_signal(symbol, prices, volume)
|
|
if not signal:
|
|
return {"status": "NO_SIGNAL", "symbol": symbol}
|
|
|
|
approved, reason = self.risk_guard.validate_order(
|
|
signal,
|
|
self.current_drawdown_pct,
|
|
len(self.open_positions)
|
|
)
|
|
|
|
if not approved:
|
|
return {"status": "REJECTED_BY_RISK", "reason": reason, "signal": signal}
|
|
|
|
# Simulated Execution
|
|
order_receipt = {
|
|
"status": "EXECUTED",
|
|
"symbol": signal["symbol"],
|
|
"action": signal["action"],
|
|
"fill_price": signal["price"],
|
|
"strategy": signal["strategy"],
|
|
"risk_status": "APPROVED"
|
|
}
|
|
self.open_positions.append(order_receipt)
|
|
return order_receipt
|
|
|
|
if __name__ == "__main__":
|
|
print("[APEX ENGINE] Liveness check passed. Engine ready.")
|