import os import json import requests from typing import Dict, Any, List, Optional import time class DeepSeekAPI: """DeepSeek API交互类，用于进行市场分析和预测""" def __init__(self, api_key: str, model: str = "deepseek-moe-16b-chat"): """ 初始化DeepSeek API Args: api_key: DeepSeek API密钥 model: 使用的模型名称 """ self.api_key = api_key self.model = model self.base_url = "https://api.deepseek.com/v1" self.headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" } def analyze_market_data(self, market_data: Dict[str, Any]) -> Dict[str, Any]: """ 分析市场数据 Args: market_data: 包含市场数据的字典，例如价格、交易量等 Returns: 分析结果 """ # 将市场数据格式化为适合大模型的格式 formatted_data = self._format_market_data(market_data) # 构建提示词 prompt = self._build_market_analysis_prompt(formatted_data) # 调用API获取分析 response = self._call_api(prompt) # 解析响应 return self._parse_analysis_response(response) def predict_price_trend(self, symbol: str, historical_data: Dict[str, Any]) -> Dict[str, Any]: """ 预测价格趋势 Args: symbol: 交易对符号，例如 'BTCUSDT' historical_data: 历史数据 Returns: 预测结果 """ # 格式化历史数据 formatted_data = self._format_historical_data(symbol, historical_data) # 构建提示词 prompt = self._build_price_prediction_prompt(symbol, formatted_data) # 调用API获取预测 response = self._call_api(prompt) # 解析响应 return self._parse_prediction_response(response) def generate_trading_strategy(self, symbol: str, analysis_result: Dict[str, Any], risk_level: str) -> Dict[str, Any]: """ 生成交易策略 Args: symbol: 交易对符号，例如 'BTCUSDT' analysis_result: 分析结果 risk_level: 风险等级，'low', 'medium', 'high' Returns: 交易策略 """ # 构建提示词 prompt = self._build_trading_strategy_prompt(symbol, analysis_result, risk_level) # 调用API获取策略 response = self._call_api(prompt) # 解析响应 return self._parse_strategy_response(response) def _call_api(self, prompt: str) -> Dict[str, Any]: """ 调用DeepSeek API Args: prompt: 提示词 Returns: API响应 """ try: endpoint = f"{self.base_url}/chat/completions" payload = { "model": self.model, "messages": [ {"role": "system", "content": "你是一个专业的加密货币分析助手，擅长分析市场趋势、预测价格走向和提供交易建议。请始终使用中文回复，并确保输出格式规范的JSON。"}, {"role": "user", "content": prompt} ], "temperature": 0.2, # 低温度使输出更加确定性 "max_tokens": 2000 } response = requests.post(endpoint, headers=self.headers, json=payload) response.raise_for_status() return response.json() except Exception as e: print(f"调用DeepSeek API时出错: {e}") return {} def _format_market_data(self, market_data: Dict[str, Any]) -> str: """ 格式化市场数据为适合大模型的格式 Args: market_data: 市场数据 Returns: 格式化的数据字符串 """ # 这里可以根据实际情况调整格式化方式 return json.dumps(market_data, indent=2) def _format_historical_data(self, symbol: str, historical_data: Dict[str, Any]) -> str: """ 格式化历史数据为适合大模型的格式 Args: symbol: 交易对符号 historical_data: 历史数据 Returns: 格式化的数据字符串 """ # 可以根据实际情况调整格式化方式 return json.dumps(historical_data, indent=2) def _build_market_analysis_prompt(self, formatted_data: str) -> str: """ 构建市场分析提示词 Args: formatted_data: 格式化的市场数据 Returns: 提示词 """ return f"""请分析以下加密货币市场数据，并提供详细的市场分析。请使用中文回复。 数据: {formatted_data} 请包括以下内容: 1. 市场总体趋势 2. 主要支撑位和阻力位 3. 交易量分析 4. 市场情绪评估 5. 关键技术指标解读（如RSI、MACD等） 请以JSON格式回复，包含以下字段: - market_trend: 市场趋势 (牛市, 熊市, 震荡) - support_levels: 支撑位列表 - resistance_levels: 阻力位列表 - volume_analysis: 交易量分析 - market_sentiment: 市场情绪 - technical_indicators: 技术指标分析 - summary: 总结 请确保回复为有效的JSON格式，并使用中文进行分析。""" def _build_price_prediction_prompt(self, symbol: str, formatted_data: str) -> str: """ 构建价格预测提示词 Args: symbol: 交易对符号 formatted_data: 格式化的历史数据 Returns: 提示词 """ return f"""请基于以下{symbol}的历史数据，预测未来24小时、7天和30天的价格走势。请使用中文回复。 历史数据: {formatted_data} 请考虑市场趋势、技术指标、历史模式和当前市场情况，提供详细的预测分析。 请以JSON格式回复，包含以下字段: - symbol: 交易对符号 - current_price: 当前价格 - prediction_24h: 24小时预测 (包含 price_range价格区间, trend趋势, confidence置信度) - prediction_7d: 7天预测 (包含 price_range价格区间, trend趋势, confidence置信度) - prediction_30d: 30天预测 (包含 price_range价格区间, trend趋势, confidence置信度) - key_factors: 影响预测的关键因素 - risk_assessment: 风险评估 请确保回复为有效的JSON格式，并使用中文进行分析。""" def _build_trading_strategy_prompt(self, symbol: str, analysis_result: Dict[str, Any], risk_level: str) -> str: """ 构建交易策略提示词 Args: symbol: 交易对符号 analysis_result: 分析结果 risk_level: 风险等级 Returns: 提示词 """ analysis_json = json.dumps(analysis_result, indent=2) return f"""请基于以下{symbol}的市场分析结果，生成一个风险等级为{risk_level}的交易策略。请使用中文回复。 分析结果: {analysis_json} 请考虑市场趋势、技术指标、风险等级和当前市场情况，提供详细的交易策略。 请以JSON格式回复，包含以下字段: - symbol: 交易对符号 - risk_level: 风险等级 (low低风险, medium中风险, high高风险) - position: 建议仓位 (买入、卖出、持有) - entry_points: 入场点列表 - exit_points: 出场点列表 - stop_loss: 止损位 - take_profit: 止盈位 - time_frame: 建议的交易时间框架 - strategy_type: 策略类型 (例如：趋势跟踪、反转、突破等) - reasoning: 策略推理过程 请确保回复为有效的JSON格式，并使用中文进行分析。""" def _parse_analysis_response(self, response: Dict[str, Any]) -> Dict[str, Any]: """ 解析分析响应 Args: response: API响应 Returns: 解析后的分析结果 """ try: if 'choices' in response and len(response['choices']) > 0: content = response['choices'][0]['message']['content'] # 尝试从响应中提取JSON start_idx = content.find('{') end_idx = content.rfind('}') + 1 if start_idx != -1 and end_idx != -1: json_str = content[start_idx:end_idx] return json.loads(json_str) return {"error": "无法从响应中提取JSON", "raw_content": content} return {"error": "API响应格式不正确", "raw_response": response} except Exception as e: print(f"解析分析响应时出错: {e}") return {"error": str(e), "raw_response": response} def _parse_prediction_response(self, response: Dict[str, Any]) -> Dict[str, Any]: """ 解析预测响应 Args: response: API响应 Returns: 解析后的预测结果 """ # 与_parse_analysis_response相同的实现 return self._parse_analysis_response(response) def _parse_strategy_response(self, response: Dict[str, Any]) -> Dict[str, Any]: """ 解析策略响应 Args: response: API响应 Returns: 解析后的策略结果 """ # 与_parse_analysis_response相同的实现 return self._parse_analysis_response(response)