本帖最后由 isaac 于 2025-1-28 14:42 编辑
import json
import requests
from datetime import datetime, timedelta
from textblob import TextBlob # 用于情感分析
import akshare as ak
import pandas as pd
import talib # 用于计算技术指标和K线形态
import streamlit as st
import plotly.graph_objs as go
from plotly.subplots import make_subplots
# 辅助函数
def validate_stock_code(symbol):
"""验证股票代码是否有效。"""
if not symbol.isdigit() or len(symbol) != 6:
return False
return True
def get_stock_basic_info(symbol):
"""获取股票基本信息。"""
stock_info = ak.stock_individual_info_em(symbol=symbol)
return stock_info
def calculate_technical_indicators(kline_data):
"""计算技术指标。"""
kline_data['MA_5' = talib.SMA(kline_data['收盘'], timeperiod=5) # 5日均线
kline_data['MA_10' = talib.SMA(kline_data['收盘'], timeperiod=10) # 10日均线
kline_data['MA_20' = talib.SMA(kline_data['收盘'], timeperiod=20) # 20日均线
kline_data['RSI' = talib.RSI(kline_data['收盘'], timeperiod=14) # 14日RSI
kline_data['MACD'], kline_data['MACD_signal'], kline_data['MACD_hist' = talib.MACD(
kline_data['收盘'], fastperiod=12, slowperiod=26, signalperiod=9
)
kline_data['VWAP' = (kline_data['收盘' * kline_data['成交量']).cumsum() / kline_data['成交量'].cumsum()
kline_data['Upper_Band'], kline_data['Middle_Band'], kline_data['Lower_Band' = talib.BBANDS(
kline_data['收盘'], timeperiod=20, nbdevup=2, nbdevdn=2, matype=0
)
kline_data['ATR' = talib.ATR(
kline_data['最高'], kline_data['最低'], kline_data['收盘'], timeperiod=14
)
kline_data['Donchian_High' = kline_data['最高'].rolling(window=20).max() # 20日最高
kline_data['Donchian_Low' = kline_data['最低'].rolling(window=20).min() # 20日最低
return kline_data
def calculate_grid_levels(kline_data):
"""计算大中小网格位。"""
high = kline_data['最高'].max() # 最高价
low = kline_data['最低'].min() # 最低价
price_range = high - low # 价格区间
large_grid = {'大网格_50%': low + price_range * 0.5}
medium_grid = {
'中网格_33.33%': low + price_range * 0.3333,
'中网格_66.67%': low + price_range * 0.6667
}
small_grid = {
'小网格_25%': low + price_range * 0.25,
'小网格_50%': low + price_range * 0.5,
'小网格_75%': low + price_range * 0.75
}
return large_grid, medium_grid, small_grid
def calculate_fibonacci_levels(kline_data):
"""计算黄金分割位。"""
high = kline_data['最高'].max() # 最高价
low = kline_data['最低'].min() # 最低价
price_range = high - low # 价格区间
fibonacci_levels = {
'黄金分割_23.6%': high - price_range * 0.236,
'黄金分割_38.2%': high - price_range * 0.382,
'黄金分割_50%': high - price_range * 0.5,
'黄金分割_61.8%': high - price_range * 0.618,
'黄金分割_78.6%': high - price_range * 0.786
}
return fibonacci_levels
def calculate_pivot_points(kline_data):
"""计算枢轴位。"""
prev_day = kline_data.iloc[-1
high = prev_day['最高'
low = prev_day['最低'
close = prev_day['收盘'
pivot_point = (high + low + close) / 3
pivot_levels = {
'枢轴点': pivot_point,
'阻力位1': 2 * pivot_point - low,
'支撑位1': 2 * pivot_point - high,
'阻力位2': pivot_point + (high - low),
'支撑位2': pivot_point - (high - low)
}
return pivot_levels
def calculate_volume_and_price_levels(kline_data):
"""计算成交量、天量、地量、天价、低价。"""
today_volume = kline_data['成交量'].iloc[-1
volume_105_high = kline_data['成交量'].rolling(window=105).max().iloc[-1
volume_105_low = kline_data['成交量'].rolling(window=105).min().iloc[-1
price_120_high = kline_data['最高'].rolling(window=120).max().iloc[-1
price_120_low = kline_data['最低'].rolling(window=120).min().iloc[-1
periods = [30, 60, 90, 120, 255
volume_and_price_levels = {}
for period in periods:
volume_and_price_levels[f'{period}日天量' = kline_data['成交量'].rolling(window=period).max().iloc[-1
volume_and_price_levels[f'{period}日地量' = kline_data['成交量'].rolling(window=period).min().iloc[-1
volume_and_price_levels[f'{period}日天价' = kline_data['最高'].rolling(window=period).max().iloc[-1
volume_and_price_levels[f'{period}日地价' = kline_data['最低'].rolling(window=period).min().iloc[-1
return {
'今日成交量': today_volume,
'105日天量': volume_105_high,
'105日地量': volume_105_low,
'120日天价': price_120_high,
'120日低价': price_120_low,
**volume_and_price_levels
}
def scan_candlestick_patterns(kline_data):
"""扫描最近5天的K线形态。"""
pattern_functions = {
'CDL2CROWS': '两只乌鸦',
'CDL3BLACKCROWS': '三只乌鸦',
'CDL3INSIDE': '三内部上涨/下跌',
'CDL3LINESTRIKE': '三线打击',
'CDL3OUTSIDE': '三外部上涨/下跌',
'CDL3STARSINSOUTH': '南方三星',
'CDL3WHITESOLDIERS': '三个白兵',
'CDLABANDONEDBABY': '弃婴',
'CDLADVANCEBLOCK': '推进块',
'CDLBELTHOLD': '腰带持有',
'CDLBREAKAWAY': '脱离',
'CDLCLOSINGMARUBOZU': '收盘光头光脚',
'CDLCONCEALBABYSWALL': '隐藏婴儿吞没',
'CDLCOUNTERATTACK': '反击线',
'CDLDARKCLOUDCOVER': '乌云盖顶',
'CDLDOJI': '十字星',
'CDLDOJISTAR': '十字星',
'CDLDRAGONFLYDOJI': '蜻蜓十字星',
'CDLENGULFING': '吞没形态',
'CDLEVENINGDOJISTAR': '黄昏十字星',
'CDLEVENINGSTAR': '黄昏之星',
'CDLGAPSIDESIDEWHITE': '侧翼白线',
'CDLGRAVESTONEDOJI': '墓碑十字星',
'CDLHAMMER': '锤子线',
'CDLHANGINGMAN': '上吊线',
'CDLHARAMI': '母子线',
'CDLHARAMICROSS': '十字母子线',
'CDLHIGHWAVE': '高浪线',
'CDLHIKKAKE': '陷阱',
'CDLHIKKAKEMOD': '修正陷阱',
'CDLHOMINGPIGEON': '归巢鸽',
'CDLIDENTICAL3CROWS': '相同三乌鸦',
'CDLINNECK': '颈内线',
'CDLINVERTEDHAMMER': '倒锤子线',
'CDLKICKING': '反冲',
'CDLKICKINGBYLENGTH': '反冲（按长度）',
'CDLLADDERBOTTOM': '梯底',
'CDLLONGLEGGEDDOJI': '长腿十字星',
'CDLLONGLINE': '长线',
'CDLMARUBOZU': '光头光脚',
'CDLMATCHINGLOW': '匹配低点',
'CDLMATHOLD': '数学持有',
'CDLMORNINGDOJISTAR': '早晨十字星',
'CDLMORNINGSTAR': '早晨之星',
'CDLONNECK': '颈上线',
'CDLPIERCING': '刺透形态',
'CDLRICKSHAWMAN': '黄包车夫',
'CDLRISEFALL3METHODS': '上升/下降三法',
'CDLSEPARATINGLINES': '分离线',
'CDLSHOOTINGSTAR': '射击之星',
'CDLSHORTLINE': '短线',
'CDLSPINNINGTOP': '纺锤顶',
'CDLSTALLEDPATTERN': '停滞形态',
'CDLSTICKSANDWICH': '三明治',
'CDLTAKURI': '探水竿',
'CDLTASUKIGAP': '跳空缺口',
'CDLTHRUSTING': '插入',
'CDLTRISTAR': '三星',
'CDLUNIQUE3RIVER': '独特三河',
'CDLUPSIDEGAP2CROWS': '向上跳空两只乌鸦',
'CDLXSIDEGAP3METHODS': '侧翼跳空三法',
}
patterns = []
for pattern_name, pattern_label in pattern_functions.items():
pattern_func = getattr(talib, pattern_name)
pattern_values = pattern_func(kline_data['开盘'], kline_data['最高'], kline_data['最低'], kline_data['收盘'])
if pattern_values.iloc[-5:].any():
patterns.append(f"{pattern_label} ({pattern_values.iloc[-1})")
return patterns
def get_latest_news(symbol):
"""获取与股票相关的最新新闻。"""
news = ak.stock_news_em(symbol=symbol)
return news
def analyze_sentiment(news):
"""对新闻进行情感分析。"""
sentiments = []
for _, row in news.iterrows():
text = row['新闻标题' + " " + row['新闻内容'
blob = TextBlob(text)
sentiment = blob.sentiment.polarity
sentiments.append(sentiment)
news['情感得分' = sentiments
return news
def generate_investment_advice_with_deepseek(symbol, basic_info, kline_data, news, patterns, grid_levels, fibonacci_levels, pivot_levels, volume_and_price_levels):
"""使用 DeepSeek-R1 生成投资建议。"""
prompt = f"""
股票基本信息：
- 公司名称：{basic_info.get('公司名称', '未知')}
- 行业：{basic_info.get('行业', '未知')}
- 市值：{basic_info.get('市值', '未知')}
K线图数据（最近5个交易日）：
{kline_data.tail(5).to_string()}
技术指标（最近5个交易日）：
- 5日均线：{kline_data['MA_5'].tail(5).to_string()}
- 10日均线：{kline_data['MA_10'].tail(5).to_string()}
- 20日均线：{kline_data['MA_20'].tail(5).to_string()}
- RSI：{kline_data['RSI'].tail(5).to_string()}
- MACD：{kline_data['MACD'].tail(5).to_string()}
- MACD Signal：{kline_data['MACD_signal'].tail(5).to_string()}
- MACD Histogram：{kline_data['MACD_hist'].tail(5).to_string()}
- VWAP：{kline_data['VWAP'].tail(5).to_string()}
- 布林带上轨：{kline_data['Upper_Band'].tail(5).to_string()}
- 布林带中轨：{kline_data['Middle_Band'].tail(5).to_string()}
- 布林带下轨：{kline_data['Lower_Band'].tail(5).to_string()}
- ATR：{kline_data['ATR'].tail(5).to_string()}
- 唐奇安上轨：{kline_data['Donchian_High'].tail(5).to_string()}
- 唐奇安下轨：{kline_data['Donchian_Low'].tail(5).to_string()}
当天K线形态：
{", ".join(patterns) if patterns else "无显著形态"}
网格位：
- 大网格位：{grid_levels[0}
- 中网格位：{grid_levels[1}
- 小网格位：{grid_levels[2}
黄金分割位：
{fibonacci_levels}
枢轴位：
{pivot_levels}
成交量与价格水平：
- 今日成交量：{volume_and_price_levels['今日成交量'}
- 105日天量：{volume_and_price_levels['105日天量'}
- 105日地量：{volume_and_price_levels['105日地量'}
- 120日天价：{volume_and_price_levels['120日天价'}
- 120日低价：{volume_and_price_levels['120日低价'}
- 30日天量：{volume_and_price_levels['30日天量'}
- 30日地量：{volume_and_price_levels['30日地量'}
- 30日天价：{volume_and_price_levels['30日天价'}
- 30日地价：{volume_and_price_levels['30日地价'}
- 60日天量：{volume_and_price_levels['60日天量'}
- 60日地量：{volume_and_price_levels['60日地量'}
- 60日天价：{volume_and_price_levels['60日天价'}
- 60日地价：{volume_and_price_levels['60日地价'}
- 90日天量：{volume_and_price_levels['90日天量'}
- 90日地量：{volume_and_price_levels['90日地量'}
- 90日天价：{volume_and_price_levels['90日天价'}
- 90日地价：{volume_and_price_levels['90日地价'}
- 120日天量：{volume_and_price_levels['120日天量'}
- 120日地量：{volume_and_price_levels['120日地量'}
- 120日天价：{volume_and_price_levels['120日天价'}
- 120日地价：{volume_and_price_levels['120日地价'}
- 255日天量：{volume_and_price_levels['255日天量'}
- 255日地量：{volume_and_price_levels['255日地量'}
- 255日天价：{volume_and_price_levels['255日天价'}
- 255日地价：{volume_and_price_levels['255日地价'}
最新新闻（情感分析）：
{news.head(5).to_string()}
请根据以上信息，生成一份详细的投资建议，包括：
1. 对股票技术指标的分析。
2. 对K线形态的分析。
3. 对网格位、黄金分割位和枢轴位的分析。
4. 对成交量与价格水平的分析。
5. 对新闻情感的分析。
6. 对股票未来走势的预测。
7. 具体的操作建议（买入、持有、卖出）。
"""
with open(f"prompt_{symbol}.txt", "w", encoding="utf-8") as file:
file.write(prompt)
payload = {
"model": "deepseek-r1:1.5b",
"prompt": prompt,
"stream": False
}
response = requests.post("http://localhost:11434/api/generate", json=payload)
if response.status_code == 200:
advice = response.json()['response'
return advice
else:
return f"请求失败，状态码：{response.status_code}，错误信息：{response.text}"
def plot_kline(symbol):
"""获取K线图数据并绘制，分析时间为过去2年到今日。"""
# 计算过去2年的日期
end_date = datetime.now().strftime("%Y%m%d") # 今日日期
start_date = (datetime.now() - timedelta(days=2 * 365)).strftime("%Y%m%d") # 过去2年的日期
# 获取K线图数据
stock_kline = ak.stock_zh_a_hist(
symbol=symbol, period="daily", start_date=start_date, end_date=end_date
)
stock_kline['date' = pd.to_datetime(stock_kline['日期'])
stock_kline.set_index('date', inplace=True)
# 计算技术指标
stock_kline = calculate_technical_indicators(stock_kline)
# 创建子图
fig = make_subplots(rows=4, cols=1, shared_xaxes=True, vertical_spacing=0.1,
subplot_titles=('K线图与移动平均线', '布林带', '相对强弱指数 (RSI)', '平均真实波幅 (ATR)'))
# 绘制收盘价和移动平均线
fig.add_trace(go.Scatter(x=stock_kline.index, y=stock_kline['收盘'], mode='lines', name='收盘价', line=dict(color='black')), row=1, col=1)
fig.add_trace(go.Scatter(x=stock_kline.index, y=stock_kline['MA_5'], mode='lines', name='5日均线', line=dict(color='blue')), row=1, col=1)
fig.add_trace(go.Scatter(x=stock_kline.index, y=stock_kline['MA_10'], mode='lines', name='10日均线', line=dict(color='green')), row=1, col=1)
fig.add_trace(go.Scatter(x=stock_kline.index, y=stock_kline['MA_20'], mode='lines', name='20日均线', line=dict(color='red')), row=1, col=1)
fig.add_trace(go.Scatter(x=stock_kline.index, y=stock_kline['VWAP'], mode='lines', name='VWAP', line=dict(color='purple')), row=1, col=1)
# 绘制布林带
fig.add_trace(go.Scatter(x=stock_kline.index, y=stock_kline['收盘'], mode='lines', name='收盘价', line=dict(color='black')), row=2, col=1)
fig.add_trace(go.Scatter(x=stock_kline.index, y=stock_kline['Upper_Band'], mode='lines', name='上轨', line=dict(color='red')), row=2, col=1)
fig.add_trace(go.Scatter(x=stock_kline.index, y=stock_kline['Middle_Band'], mode='lines', name='中轨', line=dict(color='blue')), row=2, col=1)
fig.add_trace(go.Scatter(x=stock_kline.index, y=stock_kline['Lower_Band'], mode='lines', name='下轨', line=dict(color='green')), row=2, col=1)
# 绘制RSI
fig.add_trace(go.Scatter(x=stock_kline.index, y=stock_kline['RSI'], mode='lines', name='RSI', line=dict(color='orange')), row=3, col=1)
fig.add_hline(y=70, line_dash="dash", line_color="red", annotation_text="超买 (70)", row=3, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", annotation_text="超卖 (30)", row=3, col=1)
# 绘制ATR
fig.add_trace(go.Scatter(x=stock_kline.index, y=stock_kline['ATR'], mode='lines', name='ATR', line=dict(color='blue')), row=4, col=1)
# 更新布局
fig.update_layout(
font=dict(family="SimHei", size=12, color="black"), # 设置字体为黑体
height=1200,
width=1000,
title_text=f'{symbol} K线图与技术指标（{start_date} 至 {end_date}）',
showlegend=True
)
fig.update_xaxes(title_text='日期', row=4, col=1)
fig.update_yaxes(title_text='价格', row=1, col=1)
fig.update_yaxes(title_text='价格', row=2, col=1)
fig.update_yaxes(title_text='RSI', row=3, col=1)
fig.update_yaxes(title_text='ATR', row=4, col=1)
# 显示图表
st.plotly_chart(fig)
return stock_kline
# Streamlit 应用
st.title("股票分析仪表盘")
# 输入股票代码
symbol = st.text_input("请输入股票代码（例如：601658 代表邮储银行）：", value="601658")
if st.button("分析"):
if not validate_stock_code(symbol):
st.error("股票代码无效")
else:
# 获取股票基本信息
try:
basic_info = get_stock_basic_info(symbol)
except KeyError:
st.error("股票代码无效或未找到")
st.stop()
# 获取K线图数据并绘制
kline_data = plot_kline(symbol)
# 扫描当天的K线形态
patterns = scan_candlestick_patterns(kline_data)
# 计算网格位
large_grid, medium_grid, small_grid = calculate_grid_levels(kline_data)
# 计算黄金分割位
fibonacci_levels = calculate_fibonacci_levels(kline_data)
# 计算枢轴位
pivot_levels = calculate_pivot_points(kline_data)
# 计算成交量与价格水平
volume_and_price_levels = calculate_volume_and_price_levels(kline_data)
# 获取最新新闻
news = get_latest_news(symbol)
# 对新闻进行情感分析
news = analyze_sentiment(news)
# 使用 DeepSeek-R1 生成投资建议
advice = generate_investment_advice_with_deepseek(
symbol, basic_info, kline_data, news, patterns, (large_grid, medium_grid, small_grid), fibonacci_levels, pivot_levels, volume_and_price_levels
)
# 显示技术指标
st.subheader("技术指标")
st.write(kline_data.tail(5))
# 显示K线形态
st.subheader("K线形态")
st.write(", ".join(patterns) if patterns else "无显著形态")
# 显示网格位
st.subheader("网格位")
st.write(f"大网格位: {large_grid}\n中网格位: {medium_grid}\n小网格位: {small_grid}")
# 显示黄金分割位
st.subheader("黄金分割位")
st.write(fibonacci_levels)
# 显示枢轴位
st.subheader("枢轴位")
st.write(pivot_levels)
# 显示成交量与价格水平
st.subheader("成交量与价格水平")
st.write(volume_and_price_levels)
# 显示最新新闻
st.subheader("最新新闻")
st.write(news.head(5))
# 显示新闻情感分析
st.subheader("新闻情感分析")
st.write(news[['新闻标题', '情感得分']].head(5))
# 显示投资建议
st.subheader("投资建议")
st.write(advice)
|
发表于 2025-1-28 13:38
QQ好友和群
QQ空间
腾讯微博
腾讯朋友
收藏
楼主
