本来是写给家里长辈炒股用的小玩具，既然有佬友想要就开源出来吧。
随手搓的，代码质量比较糟糕,不要嘲笑我 
项目链接ashare-llm-analyst
效果预览图
132 Likes
顺便求个star 
2 Likes
所以有帮助吗，赚了多少 
2 Likes
拿去给家里长辈用的，我不了解 
2 Likes
star了
2 Likes
感谢佬友支持
1 Like
刚看了眼main.py里，很多. ,中文标点，是敲错了吗？还是上传的时候出了问题？
4 Likes
有用吗，有没有大佬实践一些的，真是亏麻了已经 
3 Likes
嘶，我看看
1 Like
已star，感谢佬友分享
代码里确实不少中文字符
3 Likes
推荐你用cline，把代码都过一遍
4 Likes
找到原因了，我在网页端编辑了一次main的内容，然后代码就被浏览器插件污染了
4 Likes
感谢佬友分享~
3 Likes
感谢佬友的分享~
3 Likes
学习了大佬
3 Likes
修改几个地方可以兼容openai格式：
main.py class StockAnalyzer 的初始化函数：
def __init__(
self,
_stock_info,
deepseek_api_key,
deepseek_base_url,
model,
count=120,
):
"""
初始化股票分析器
Args:
_stock_info: 股票信息字典
count: 获取的数据条数
deepseek_api_key: Deepseek API密钥
deepseek_base_url: Deepseek API基础URL
"""
self.stock_codes = list(_stock_info.values())
self.stock_names = _stock_info
self.count = count
self.data = {}
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False
# 初始化Deepseek分析器
self.deepseek = (
DeepseekAnalyzer(deepseek_api_key, deepseek_base_url, model=model)
if deepseek_api_key
else None
)
Deepseek.py 的 class DeepseekAnalyzer
class DeepseekAnalyzer:
"""使用 OpenAI SDK 与 Deepseek API 交互的类"""
def __init__(
self,
api_key: str,
base_url: str ,
model: str ,
):
"""
初始化 Deepseek 分析器
Args:
api_key (str): Deepseek API 密钥
base_url (str): Deepseek API 基础 URL
"""
self.client = OpenAI(api_key=api_key, base_url=base_url)
self.model = model
def request_analysis(
self, df: pd.DataFrame, technical_indicators: pd.DataFrame
) -> Optional[Dict[str, Any]]:
"""
向 Deepseek API 发送分析请求
Args:
df (pd.DataFrame): 原始股票数据
technical_indicators (pd.DataFrame): 技术指标数据
Returns:
Optional[Dict[str, Any]]: API 响应的分析结果
"""
try:
# 准备数据
print("开始准备数据...")
data_str = _format_data_for_prompt(df, technical_indicators)
print(f"数据准备完成，数据长度: {len(data_str)}")
# 构建消息
print("构建API请求消息...")
messages = [
{"role": "system", "content": _create_system_prompt()},
{
"role": "user",
"content": f"请分析以下股票数据并给出专业的分析意见：\n{data_str}",
},
]
print(f"消息构建完成，系统提示词长度: {len(messages[0]['content'])}")
print(f"用户消息长度: {len(messages[1]['content'])}")
# 发送请求
print("开始发送API请求...")
try:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=1.0,
stream=False,
max_tokens=10000,
)
print("API请求发送成功")
except Exception as api_e:
# 检查是否是空响应导致的JSON解析错误
if str(api_e).startswith("Expecting value: line 1 column 1 (char 0)"):
print("API返回空响应，服务器可能繁忙")
raise APIBusyError("API服务器繁忙，返回空响应") from api_e
print(f"API请求发送失败: {str(api_e)}")
raise # 重新抛出其他类型的异常
# 记录原始响应以便调试
print("API 原始响应类型:", type(response))
print("API 原始响应内容:", response)
# 检查响应内容
if not response:
print("API返回空响应")
return format_analysis_result({})
if not hasattr(response, "choices"):
print(f"API响应缺少choices属性，响应结构: {dir(response)}")
return format_analysis_result({})
if not response.choices:
print("API响应的choices为空")
return format_analysis_result({})
# 解析响应
try:
analysis_text = response.choices[0].message.content
print("成功获取分析文本内容")
print("分析文本:", analysis_text)
except Exception as text_e:
print(f"获取分析文本失败: {str(text_e)}")
raise
# 将文本响应组织成结构化数据
print("开始解析分析文本...")
result = _parse_analysis_response(analysis_text)
print("分析文本解析完成")
return result
except APIBusyError as be: # 处理API繁忙异常
print(f"=== API繁忙错误 ===")
print(f"错误详情: {str(be)}")
print(f"错误类型: {type(be)}")
return format_analysis_result({})
except json.JSONDecodeError as je:
print(f"=== JSON解析错误 ===")
print(f"错误详情: {str(je)}")
print(f"错误类型: {type(je)}")
print(f"错误位置: {je.pos}")
print(f"错误行列: 行 {je.lineno}, 列 {je.colno}")
print(f"错误的文档片段: {je.doc[:100] if je.doc else 'None'}")
return format_analysis_result({})
except openai.APITimeoutError as te:
print(f"=== API超时错误 ===")
print(f"错误详情: {str(te)}")
print(f"错误类型: {type(te)}")
return format_analysis_result({})
except openai.APIConnectionError as ce:
print(f"=== API连接错误 ===")
print(f"错误详情: {str(ce)}")
print(f"错误类型: {type(ce)}")
return format_analysis_result({})
except openai.APIError as ae:
print(f"=== API错误 ===")
print(f"错误详情: {str(ae)}")
print(f"错误类型: {type(ae)}")
return format_analysis_result({})
except openai.RateLimitError as re:
print(f"=== API频率限制错误 ===")
print(f"错误详情: {str(re)}")
print(f"错误类型: {type(re)}")
return format_analysis_result({})
except Exception as e:
print(f"=== 未预期的错误 ===")
print(f"错误详情: {str(e)}")
print(f"错误类型: {type(e)}")
print(f"错误追踪:")
import traceback
traceback.print_exc()
return format_analysis_result({})
修改后的调用示例（在main.py的最下面）
if __name__ == "__main__":
stock_info = {"上证指数": "sh000001"}
base_url = "https://hq.sinajs.cn/v1" #一定要包含v1
model = "gemini-2.0-flash"
api_key = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
analyzer = StockAnalyzer(stock_info, api_key, base_url, model)
report_path = analyzer.run_analysis()
print(f"分析报告已生成: {report_path}")
13 Likes
感谢大佬！
6 Likes
挺好的，一开始我就是用 openai的SDK去对接的，但是DS在繁忙时会持续返回空行来维持TCP连接，OAI的SDK捕获不到这种情况。折腾完catch error就懒得去调兼容了
3 Likes
建议改一个地方，数据不用直接生成html，而是存入到一个本地数据库里，然后再生成一个查询系统，可以用streamlit实现，去调用数据库的查询，数据的拖拽和查看更方便，而且可以设置多页，也可以多数据多维度对比。
9 Likes