demand_server/app/strategies/registry.py

import threading
from dataclasses import dataclass
from typing import Any

from app.strategies.base import BaseStrategy
from app.strategies.config_store import StrategyConfigRecord, fetch_all_configs, fetch_config_by_id


@dataclass(frozen=True)
class ActiveStrategy:
    strategy: BaseStrategy
    config: StrategyConfigRecord


class StrategyRegistry:
    """策略注册表：代码实现与 DB 配置解耦，支持热更新。"""

    _registry: dict[str, BaseStrategy] = {}
    _config_cache: dict[str, StrategyConfigRecord] = {}
    _lock = threading.RLock()

    @classmethod
    def register(cls, strategy: BaseStrategy) -> None:
        strategy_id = strategy.strategy_id.strip()
        if not strategy_id:
            raise ValueError("strategy_id is required")

        with cls._lock:
            if strategy_id in cls._registry:
                raise ValueError(f"strategy already registered: {strategy_id}")

            cached = cls._config_cache.get(strategy_id)
            if cached is not None and not strategy.validate_config(cached.params):
                raise ValueError(f"invalid config for strategy: {strategy_id}")

            cls._registry[strategy_id] = strategy

    @classmethod
    def get(cls, strategy_id: str) -> BaseStrategy | None:
        with cls._lock:
            return cls._registry.get(strategy_id)

    @classmethod
    def load_all_configs(cls) -> int:
        records = fetch_all_configs()
        with cls._lock:
            cls._config_cache = {record.strategy_id: record for record in records}
        return len(records)

    @classmethod
    def get_active(cls) -> list[ActiveStrategy]:
        with cls._lock:
            active: list[ActiveStrategy] = []
            for strategy_id, strategy in cls._registry.items():
                config = cls._config_cache.get(strategy_id)
                if config is None or not config.active:
                    continue
                active.append(ActiveStrategy(strategy=strategy, config=config))
            return active

    @classmethod
    def get_unregistered_active_configs(cls) -> list[StrategyConfigRecord]:
        """DB 中 active=true 但未注册代码实现的策略。"""
        with cls._lock:
            return [
                config
                for strategy_id, config in cls._config_cache.items()
                if config.active and strategy_id not in cls._registry
            ]

    @classmethod
    def reload_config(cls, strategy_id: str) -> StrategyConfigRecord:
        record = fetch_config_by_id(strategy_id)
        if record is None:
            raise KeyError(f"strategy config not found: {strategy_id}")

        with cls._lock:
            strategy = cls._registry.get(strategy_id)
            if strategy is not None and not strategy.validate_config(record.params):
                raise ValueError(f"invalid config for strategy: {strategy_id}")
            cls._config_cache[strategy_id] = record
            return record

    @classmethod
    def get_config(cls, strategy_id: str) -> StrategyConfigRecord | None:
        with cls._lock:
            return cls._config_cache.get(strategy_id)

    @classmethod
    def registered_strategy_ids(cls) -> list[str]:
        with cls._lock:
            return sorted(cls._registry.keys())

    @classmethod
    def list_registered(cls) -> list[dict[str, str]]:
        with cls._lock:
            return [
                {
                    "strategy_id": strategy.strategy_id,
                    "name": strategy.name,
                    "version": strategy.version,
                }
                for strategy in sorted(
                    cls._registry.values(),
                    key=lambda item: item.strategy_id,
                )
            ]

    @classmethod
    def validate_config_for_strategy(cls, strategy_id: str, config: dict[str, Any]) -> bool:
        with cls._lock:
            strategy = cls._registry.get(strategy_id)
        if strategy is None:
            raise KeyError(f"strategy not registered: {strategy_id}")
        return strategy.validate_config(config)