DualFlow/algorithm/base/algorithm_base.py

from abc import ABC, abstractmethod

from typing import Dict

class AlgorithmBase(ABC):
    """算法接入标准"""

    def __init__(self, params=None):
        """算法初始化"""
        pass

    def __call__(self, x:Dict, *args, **kwargs)->Dict:
        """调用主逻辑，子类必须实现"""
        pass

    @abstractmethod
    def get(self):
        """获取算法实时运行参数，子类必须实现"""
        pass

    @abstractmethod
    def set(self):
        """实现参数热更新，子类必须实现"""
        pass