bot/objects/DataModel.py - dev - fgtmenow/tb

bot/objects/DataModel.py
145 строк6 KB
Azarev Artem
start implement bot
15 апр 2025, 16:52
15 апр 2025, 16:521ac3c18
import pandas as pd
from pandas import Timestamp

from _common.domain.model.Bar import Bar
from analyze.incremnt import resample_ohlcv_data
from bot.data.AnalyzeRepository import apply_tech_analyze


class DataModel:
    """Класс для управления данными финансового рынка с различными таймфреймами и техническими индикаторами."""

    def __init__(self):
        self.data_1m: dict[str, pd.DataFrame] = {}
        self.data_5m: dict[str, pd.DataFrame] = {}
        # self.max_length = 10080

    def set_initial(self, symbol: str, df: pd.DataFrame):
        """
        Установить начальные данные для символа.
        
        Аргументы:
            symbol: Символ торговой пары или актива
            df: DataFrame с историческими данными
        """
        self.data_1m[symbol] = df
        self.data_5m[symbol] = resample_ohlcv_data(self.data_1m[symbol], "5min")
        apply_tech_analyze(self.data_1m[symbol], self.data_5m[symbol])

    def append_timeframe(self,
                         symbol: str,
                         open: float,
                         high: float,
                         low: float,
                         close: float,
                         volume: float,
                         timestamp: Timestamp
                         ):
        """
        Добавить новые рыночные данные для заданного символа.
        
        Аргументы:
            symbol: Символ торговой пары или актива
            open: Цена открытия
            high: Максимальная цена
            low: Минимальная цена
            close: Цена закрытия
            volume: Объем торгов
            timestamp: Unix-время
        """
        new_data = [timestamp, open, high, low, close, volume]
        new_data = pd.DataFrame([new_data], columns=[
            'timestamp',
            'open',
            'high',
            'low',
            'close',
            'volume'
        ], )

        if self.data_1m.__contains__(symbol):
            if timestamp in self.data_1m[symbol].index:
                print(self.data_1m[symbol].index)
                return
            self.data_1m[symbol] = pd.concat([self.data_1m[symbol], new_data])
            self.data_1m[symbol].index = pd.to_datetime(self.data_1m[symbol]['timestamp'], unit='ms')
        else:
            self.data_1m[symbol] = new_data

        if len(self.data_1m[symbol]) % 5 == 0:
            self.data_5m[symbol] = resample_ohlcv_data(self.data_1m[symbol], "5min")
            self.data_1m[symbol] = self.data_1m[symbol].tail(len(self.data_1m[symbol]) - 5)
        if self.data_5m.__contains__(symbol):
            apply_tech_analyze(self.data_1m[symbol], self.data_5m[symbol])

    def get(self, symbol: str, key: str, position: int = 0):
        """
        Получить конкретное значение данных для символа.
        
        Аргументы:
            symbol: Символ торговой пары или актива
            key: Название столбца данных
            position: Позиция от конца набора данных (0 или отрицательное для последних данных)
        """
        return self.data_1m[symbol][key].iloc[position - 1]

    def check_ready(self, symbol: str) -> bool:
        """
        Проверить, достаточно ли исторических данных для технического анализа.
        
        Аргументы:
            symbol: Символ торговой пары или актива
        Возвращает:
            bool: True, если достаточно точек данных
        """
        return len(self.data_1m[symbol]) > 10

    #########
    # Convenience methods for accessing specific market data fields

    def timestamp(self, symbol: str, position: int = 0):
        """Получить временную метку для определенной позиции."""
        return self.data_1m[symbol]['timestamp'].iloc[position - 1]

    def high(self, symbol: str, position: int = 0):
        """Получить максимальную цену для определенной позиции."""
        return self.data_1m[symbol]['high'].iloc[position - 1]

    def low(self, symbol: str, position: int = 0):
        """Получить минимальную цену для определенной позиции."""
        return self.data_1m[symbol]['low'].iloc[position - 1]

    def open(self, symbol: str, position: int = 0):
        """Получить цену открытия для определенной позиции."""
        return self.data_1m[symbol]['open'].iloc[position - 1]

    def volume(self, symbol: str, position: int = 0):
        """Получить объем торгов для определенной позиции."""
        return self.data_1m[symbol]['volume'].iloc[position - 1]

    def close(self, symbol: str, position: int = 0):
        """Получить цену закрытия для определенной позиции."""
        return self.data_1m[symbol]['close'].iloc[position - 1]

    def get_bar(self, symbol: str, position: int = 0) -> Bar:
        """
        Получить полный объект Bar для определенной позиции.
        
        Аргументы:
            symbol: Символ торговой пары или актива
            position: Позиция от конца набора данных
        Возвращает:
            Bar: Объект, содержащий все данные о ценах и объеме
        """
        return Bar(
            self.timestamp(symbol, position - 1),
            self.high(symbol, position - 1),
            self.low(symbol, position - 1),
            self.open(symbol, position - 1),
            self.close(symbol, position - 1),
            self.volume(symbol, position - 1)
        )

    def data_length(self, symbol: str):
        """Получить количество доступных точек данных для символа."""
        return len(self.data_1m[symbol])