''' 索普量化聚宽交易服务器2.0 作者索普量化 微信:xms_quants1 ''' from dash import html, dcc, Input, Output, dash_table, dash import pandas as pd import os from datetime import datetime from pathlib import Path import logging from typing import Dict, Any, Optional # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class TradingDataManager: """交易数据管理类""" def __init__(self, base_path: str): self.base_path = Path(base_path) self.realtime_data_dir = self.base_path / "data" / "实时数据" self.historical_data_dir = self.base_path / "实时数据" / "历史数据" self.auth_file = self.base_path / "授权表" / "授权表.xlsx" # 创建必要的目录 self.realtime_data_dir.mkdir(parents=True, exist_ok=True) self.historical_data_dir.mkdir(parents=True, exist_ok=True) self.auth_file.parent.mkdir(parents=True, exist_ok=True) def get_user_file_path(self, user_id: str, data_type: str) -> Path: """获取用户数据文件路径""" if data_type == "实时数据": return self.realtime_data_dir / f"{user_id}实时数据.csv" elif data_type == "历史数据": return self.historical_data_dir / f"{user_id}历史数据.csv" else: raise ValueError(f"不支持的数据类型: {data_type}") def load_user_data(self, user_id: str, data_type: str) -> pd.DataFrame: """加载用户数据""" file_path = self.get_user_file_path(user_id, data_type) try: if file_path.exists(): df = pd.read_csv(file_path) # 清理可能的索引列 df = df.drop(columns=['Unnamed: 0'], errors='ignore') return df return pd.DataFrame() except Exception as e: logger.error(f"加载用户数据失败: {e}") return pd.DataFrame() def save_user_data(self, user_id: str, data_type: str, df: pd.DataFrame): """保存用户数据""" file_path = self.get_user_file_path(user_id, data_type) try: df.to_csv(file_path, index=False) except Exception as e: logger.error(f"保存用户数据失败: {e}") def load_auth_data(self) -> pd.DataFrame: """加载授权数据""" try: if self.auth_file.exists(): df = pd.read_excel(self.auth_file, dtype='object') df = df.drop(columns=['Unnamed: 0'], errors='ignore') return df return pd.DataFrame(columns=['用户名称', '到期时间']) except Exception as e: logger.error(f"加载授权数据失败: {e}") return pd.DataFrame(columns=['用户名称', '到期时间']) class AuthManager: """授权管理类""" @staticmethod def calculate_days_remaining(start_date: str, end_date: str) -> int: """计算剩余天数""" try: start = datetime.strptime(start_date, '%Y-%m-%d') end = datetime.strptime(end_date, '%Y-%m-%d') return (end - start).days except Exception as e: logger.error(f"计算天数失败: {e}") return -1 @staticmethod def validate_user(auth_df: pd.DataFrame, user_id: str) -> Dict[str, Any]: """验证用户授权""" try: auth_df['用户名称'] = auth_df['用户名称'].astype(str) auth_df['到期时间'] = auth_df['到期时间'].astype(str) if user_id not in auth_df['用户名称'].values: return {"valid": False, "message": "用户没有注册请注册/联系作者微信15117320079"} user_data = auth_df[auth_df['用户名称'] == user_id].iloc[0] end_date = str(user_data['到期时间'])[:10] current_date = datetime.now().strftime('%Y-%m-%d') days_remaining = AuthManager.calculate_days_remaining(current_date, end_date) if days_remaining >= 1: return { "valid": True, "days_remaining": days_remaining, "user_data": user_data } else: return {"valid": False, "message": "授权码到期找作者获取"} except Exception as e: logger.error(f"用户验证失败: {e}") return {"valid": False, "message": "授权验证失败"} class DataProcessor: """数据处理类""" @staticmethod def parse_signal_text(text: str) -> Dict[str, Any]: """解析信号文本""" try: cleaned_text = '\n'.join([ line.strip().lstrip() for line in text.split('\n') if line.strip() ]) return eval(cleaned_text) except Exception as e: logger.error(f"解析信号文本失败: {e}") return {} @staticmethod def process_stock_data(df: pd.DataFrame) -> pd.DataFrame: """处理股票数据""" if df.empty: return df # 清理股票代码 if '股票代码' in df.columns: df['证券代码'] = df['股票代码'].apply( lambda x: str(x).split('.XSHE')[0].split('.XSHG')[0] ) df['数据长度'] = df['证券代码'].apply(lambda x: len(str(x))) df = df[df['数据长度'] >= 6] df = df.drop(columns=['数据长度'], errors='ignore') return df @staticmethod def create_signal_data(signal_dict: Dict[str, Any]) -> pd.DataFrame: """创建信号数据""" return pd.DataFrame([signal_dict]) class TradingApp: """交易应用主类""" def __init__(self): self.path = os.path.dirname(os.path.abspath(__file__)) self.data_manager = TradingDataManager(self.path) self.auth_manager = AuthManager() self.data_processor = DataProcessor() self.select_options = [ '用户信息', '发送信号', '实时数据', '历史数据', '清空实时数据', '清空历史数据' ] self.app = dash.Dash(__name__) self.setup_layout() self.setup_callbacks() def setup_layout(self): """设置应用布局""" self.app.layout = html.Div([ html.H1([html.A('索普聚宽交易使用教程2.0',href='https://gitee.com/li-xingguo11111/big_qmt_joinquant_trader',style={'textAlign': 'center'}),]), html.H3('索普量化,找作者获取授权码 微信xms_quants1,交易函数数据函数分离,平台只是一个信号的中转平台,聚宽的策略仔细研究避免未来函数,投资有风险,平台不做投资参考',style={'textAlign': 'center'}), self._create_control_table(), self._create_text_area(), html.H3('数据展示'), self._create_data_table(), dcc.Download(id='joinquant_trader_table_down') ]) def _create_control_table(self): """创建控制表格""" return html.Table([ html.Tr([ html.Td('用户名称', style={'border': '1px solid', 'width': '200px'}), html.Td('数据类型', style={'border': '1px solid', 'width': '200px'}), html.Td('运行程序', style={'border': '1px solid', 'width': '200px'}), html.Td('下载数据', style={'border': '1px solid', 'width': '200px'}) ]), html.Tr([ html.Td( dcc.Input(value='123456', id='joinquant_trader_password'), style={'border': '1px solid', 'width': '200px'} ), html.Td( dcc.Dropdown( options=self.select_options, value='用户信息', id='joinquant_trader_data_type' ), style={'border': '1px solid', 'width': '200px'} ), html.Td( dcc.Dropdown( options=['运行', '不运行'], id='joinquant_trader_run', value='不运行' ), style={'border': '1px solid', 'width': '200px'} ), html.Td( dcc.RadioItems( options={'下载数据': "下载数据", "不下载数据": "不下载数据"}, id='joinquant_trader_down_data', value='不下载数据' ), style={'border': '1px solid', 'width': '200px'} ) ]), ]) def _create_text_area(self): """创建文本区域""" return dcc.Textarea( value=""" {'状态': 'held', '订单添加时间': 'datetime.datetime(2024, 4, 23, 9, 30)', '买卖': 'False', '下单数量': '9400', '已经成交': '9400', '股票代码': '', '订单ID': '1732208241', '平均成交价格': '10.5', '持仓成本': '10.59', '多空': 'long', '交易费用': '128.31'} """, id='joinquant_trader_text', style={'width': '80%', 'height': 200, "text-align": "left"} ) def _create_data_table(self): """创建数据表格""" return dash_table.DataTable( id='joinquant_trader_table', page_size=10, style_table={'font-size': 15}, sort_action='native' ) def setup_callbacks(self): """设置回调函数""" # 主数据回调 @self.app.callback( Output('joinquant_trader_table', 'data', allow_duplicate=True), Input('joinquant_trader_password', 'value'), Input('joinquant_trader_data_type', 'value'), Input('joinquant_trader_text', 'value'), Input('joinquant_trader_run', 'value'), Input('joinquant_trader_down_data', 'value'), prevent_initial_call=True ) def update_table(password, data_type, text, run, down_data): return self._handle_data_update(password, data_type, text, run, down_data) # 下载回调 @self.app.callback( Output('joinquant_trader_table_down', 'data', allow_duplicate=True), Input('joinquant_trader_password', 'value'), Input('joinquant_trader_data_type', 'value'), Input('joinquant_trader_text', 'value'), Input('joinquant_trader_run', 'value'), Input('joinquant_trader_down_data', 'value'), prevent_initial_call=True ) def download_data(password, data_type, text, run, down_data): return self._handle_download(password, data_type, text, run, down_data) def _handle_data_update(self, password: str, data_type: str, text: str, run: str, down_data: str) -> list: """处理数据更新""" if run != '运行': return self._create_message_df("没有点击运行选择", False).to_dict('records') auth_df = self.data_manager.load_auth_data() auth_result = self.auth_manager.validate_user(auth_df, str(password)) if not auth_result["valid"]: return self._create_message_df(auth_result["message"], False).to_dict('records') return self._process_data_type( password, data_type, text, auth_result ).to_dict('records') def _process_data_type(self, user_id: str, data_type: str, text: str, auth_result: Dict[str, Any]) -> pd.DataFrame: """处理不同类型的数据请求""" current_date = datetime.now().strftime('%Y-%m-%d') if data_type == '用户信息': return self._handle_user_info(auth_result) elif data_type == '发送信号': return self._handle_send_signal(user_id, text, current_date) elif data_type in ['实时数据', '历史数据']: return self._handle_data_query(user_id, data_type) elif data_type in ['清空实时数据', '清空历史数据']: return self._handle_clear_data(user_id, data_type) else: return self._create_message_df("未知的数据类型", False) def _handle_user_info(self, auth_result: Dict[str, Any]) -> pd.DataFrame: """处理用户信息请求""" user_data = auth_result["user_data"].copy() user_data['到期天数'] = auth_result["days_remaining"] user_data['信息推送'] = '授权码正常' user_data['数据状态'] = True return pd.DataFrame([user_data]) def _handle_send_signal(self, user_id: str, text: str, current_date: str) -> pd.DataFrame: """处理发送信号""" signal_data = self.data_processor.parse_signal_text(text) if not signal_data: return self._create_message_df("信号数据解析失败", False) # 处理实时数据 realtime_df = self.data_manager.load_user_data(user_id, "实时数据") realtime_df = self.data_processor.process_stock_data(realtime_df) # 处理历史数据 historical_df = self.data_manager.load_user_data(user_id, "历史数据") historical_df = self.data_processor.process_stock_data(historical_df) # 创建新信号数据 new_signal = self.data_processor.create_signal_data(signal_data) new_signal['交易日'] = current_date new_signal['数据状态'] = True # 合并数据并去重 realtime_updated = self._merge_and_deduplicate(realtime_df, new_signal, current_date) historical_updated = self._merge_and_deduplicate(historical_df, new_signal, current_date) # 保存更新后的数据 self.data_manager.save_user_data(user_id, "实时数据", realtime_updated) self.data_manager.save_user_data(user_id, "历史数据", historical_updated) return new_signal def _handle_data_query(self, user_id: str, data_type: str) -> pd.DataFrame: """处理数据查询""" df = self.data_manager.load_user_data(user_id, data_type) df = self.data_processor.process_stock_data(df) return df if not df.empty else self._create_message_df(f"{data_type}为空", True) def _handle_clear_data(self, user_id: str, data_type: str) -> pd.DataFrame: """处理清空数据""" clear_type = "实时数据" if "实时" in data_type else "历史数据" self.data_manager.save_user_data(user_id, clear_type, pd.DataFrame()) return self._create_message_df(f"{clear_type}已清空", True) def _merge_and_deduplicate(self, existing_df: pd.DataFrame, new_df: pd.DataFrame, current_date: str) -> pd.DataFrame: """合并数据并去重""" merged_df = pd.concat([existing_df, new_df], ignore_index=True) merged_df['交易日'] = current_date # 去重逻辑 dup_cols = ['股票代码', '下单数量', '买卖', '多空'] if all(col in merged_df.columns for col in dup_cols): merged_df = merged_df.drop_duplicates(subset=dup_cols, keep='last') return merged_df def _handle_download(self, password: str, data_type: str, text: str, run: str, down_data: str): """处理数据下载""" if down_data != '下载数据': return None if data_type not in ['实时数据', '历史数据']: return None auth_df = self.data_manager.load_auth_data() auth_result = self.auth_manager.validate_user(auth_df, str(password)) if not auth_result["valid"]: return None df = self.data_manager.load_user_data(str(password), data_type) current_date = datetime.now().strftime('%Y-%m-%d') return dcc.send_data_frame( df.to_excel, filename=f'{current_date}{data_type}.xlsx' ) @staticmethod def _create_message_df(message: str, status: bool) -> pd.DataFrame: """创建消息DataFrame""" return pd.DataFrame({'信息提示': [message], '数据状态': [status]}) def run(self, host: str = '127.0.0.1', port: str = '8025'): """运行应用""" try: self.app.run(debug=True, host=host, port=port) except Exception as e: logger.error(f"启动应用失败: {e}") # 尝试备用端口 try: self.app.run_server(debug=True, host=host, port=port) except Exception as e: logger.error(f"备用端口启动也失败: {e}") if __name__ == '__main__': host = '127.0.0.1' port = '8025' app = TradingApp() app.run(host=host,port=port)