61 lines
1.7 KiB
Python
61 lines
1.7 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
统计数据服务模块
|
|
"""
|
|
|
|
from typing import Dict, Optional, List
|
|
|
|
from ..models.statistics import StatisticsData
|
|
from ..store.statistics_store import StatisticsStore
|
|
|
|
|
|
class StatisticsService:
|
|
"""
|
|
统计数据服务
|
|
|
|
功能：
|
|
1. 处理EA上报的统计数据
|
|
2. 获取品种价差
|
|
3. 获取账户信息
|
|
"""
|
|
|
|
def __init__(self, store: StatisticsStore = None):
|
|
self.store = store or StatisticsStore()
|
|
|
|
def process_statistics(self, data: Dict) -> None:
|
|
"""
|
|
处理EA上报的统计数据
|
|
|
|
Args:
|
|
data: EA上报的JSON数据
|
|
"""
|
|
stat = StatisticsData.from_ea_data(data)
|
|
self.store.add(stat)
|
|
|
|
def get_latest(self, symbol: str = None) -> Optional[StatisticsData]:
|
|
"""获取最新统计数据"""
|
|
return self.store.get_latest(symbol)
|
|
|
|
def get_spread(self, symbol: str) -> Optional[float]:
|
|
"""获取品种价差"""
|
|
return self.store.get_spread(symbol)
|
|
|
|
def get_mid_price(self, symbol: str) -> Optional[float]:
|
|
"""获取品种中间价"""
|
|
latest = self.store.get_latest(symbol)
|
|
if latest:
|
|
return latest.mid_price
|
|
return None
|
|
|
|
def get_account_info(self, symbol: str = None) -> Dict:
|
|
"""获取账户信息"""
|
|
return self.store.get_account_info(symbol)
|
|
|
|
def get_by_symbol(self, symbol: str, count: int = 10) -> List[StatisticsData]:
|
|
"""获取指定品种的统计数据"""
|
|
return self.store.get_by_symbol(symbol, count)
|
|
|
|
def get_status(self) -> Dict:
|
|
"""获取服务状态"""
|
|
return self.store.get_status() |