| import asyncio |
| import logging |
|
|
| from app.models.schemas import StrategyConfig, StrategyStatus |
| from app.services.market_data import market_data_service |
| from app.services.order_manager import order_manager |
| from app.strategies.base import BaseStrategy |
| from app.strategies.bollinger_bands import BollingerBandsStrategy |
| from app.strategies.dual_thrust import DualThrustStrategy |
| from app.strategies.ma_crossover import MACrossoverStrategy |
|
|
| logger = logging.getLogger(__name__) |
|
|
| STRATEGY_CLASSES: dict[str, type[BaseStrategy]] = { |
| "ma_crossover": MACrossoverStrategy, |
| "bollinger_bands": BollingerBandsStrategy, |
| "dual_thrust": DualThrustStrategy, |
| } |
|
|
|
|
| class StrategyEngine: |
| def __init__(self): |
| self._strategies: dict[str, BaseStrategy] = {} |
| self._configs: dict[str, StrategyConfig] = {} |
| self._tasks: dict[str, asyncio.Task] = {} |
|
|
| def get_available_strategies(self) -> list[dict]: |
| result = [] |
| for key, cls in STRATEGY_CLASSES.items(): |
| instance = cls(strategy_id="temp", symbol="") |
| result.append({ |
| "type": key, |
| "name": instance.name, |
| "description": instance.description, |
| "default_params": instance.default_params, |
| }) |
| return result |
|
|
| def get_running_strategies(self) -> list[StrategyConfig]: |
| return list(self._configs.values()) |
|
|
| def get_strategy_signals(self, strategy_id: str) -> list[dict]: |
| strategy = self._strategies.get(strategy_id) |
| if strategy: |
| return strategy.signals |
| return [] |
|
|
| def get_strategy_performance(self, strategy_id: str): |
| strategy = self._strategies.get(strategy_id) |
| if strategy: |
| return strategy.performance |
| return None |
|
|
| def add_strategy(self, config: StrategyConfig) -> StrategyConfig: |
| cls = STRATEGY_CLASSES.get(config.strategy_type) |
| if cls is None: |
| raise ValueError(f"Unknown strategy type: {config.strategy_type}") |
|
|
| strategy = cls( |
| strategy_id=config.strategy_id, |
| symbol=config.symbol, |
| params=config.params, |
| ) |
| self._strategies[config.strategy_id] = strategy |
| self._configs[config.strategy_id] = config |
| return config |
|
|
| def start_strategy(self, strategy_id: str) -> StrategyConfig | None: |
| config = self._configs.get(strategy_id) |
| if config is None: |
| return None |
| config.status = StrategyStatus.RUNNING |
| task = asyncio.create_task(self._run_strategy(strategy_id)) |
| self._tasks[strategy_id] = task |
| return config |
|
|
| def stop_strategy(self, strategy_id: str) -> StrategyConfig | None: |
| config = self._configs.get(strategy_id) |
| if config is None: |
| return None |
| config.status = StrategyStatus.STOPPED |
| task = self._tasks.pop(strategy_id, None) |
| if task: |
| task.cancel() |
| return config |
|
|
| def remove_strategy(self, strategy_id: str) -> bool: |
| self.stop_strategy(strategy_id) |
| self._strategies.pop(strategy_id, None) |
| self._configs.pop(strategy_id, None) |
| return True |
|
|
| async def _run_strategy(self, strategy_id: str): |
| strategy = self._strategies.get(strategy_id) |
| config = self._configs.get(strategy_id) |
| if not strategy or not config: |
| return |
|
|
| try: |
| while config.status == StrategyStatus.RUNNING: |
| klines = market_data_service.get_kline_history(strategy.symbol) |
| if klines: |
| signal = strategy.calculate_signal(klines) |
| if signal: |
| order = order_manager.place_order(signal) |
| logger.info( |
| "Strategy %s generated %s signal for %s, order: %s", |
| strategy_id, signal.side, signal.symbol, order.order_id |
| ) |
| await asyncio.sleep(2) |
| except asyncio.CancelledError: |
| pass |
| except Exception as e: |
| logger.error("Strategy %s error: %s", strategy_id, e) |
| config.status = StrategyStatus.ERROR |
|
|
|
|
| strategy_engine = StrategyEngine() |
|
|
