本文介绍了一种基于 Python 和 Ollama 的股票技术分析系统。系统利用 yfinance 获取市场数据，通过 pandas 计算滚动平均、EMA、RSI 及布林带等技术指标，并将这些指标输入本地运行的 Llama 3 模型，生成自然语言形式的市场趋势解读。最后，使用 Streamlit 构建可视化界面展示实时数据和 AI 分析结果。该方案实现了从数据采集、指标计算到智能分析的自动化流程，适用于量化监控场景。
本文介绍了一个结合 Python 数据分析库、本地大模型（Ollama）以及 Streamlit 可视化的股票分析系统。该系统能够实时获取股票数据，计算关键技术指标，并利用 Llama 3 模型对市场行情进行自然语言解读。
yfinance 获取 Apple (AAPL) 和道琼斯指数 (DJI) 的历史行情。在开始之前，请确保已安装以下 Python 包：
import yfinance as yf
import pandas as pd
import schedule
import time
import ollama
from datetime import datetime, timedelta
同时，确保本地已运行 Ollama 服务并拉取了 llama3 模型：
ollama pull llama3
首先，获取昨日分钟级历史数据作为模拟源。这里以苹果股票和道琼斯指数为例。
stock = yf.Ticker("AAPL")
dow_jones = yf.Ticker("^DJI")
data = stock.history(period="1d", interval="1m")
dow_data = dow_jones.history(period="1d", interval="1m")
print(data.head())
为了跟踪滚动窗口内的数据和每日市场状态，我们定义以下全局变量：
rolling_window = pd.DataFrame()
dow_rolling_window = pd.DataFrame()
# 用于跟踪每日市场情况
daily_high = float('-inf')
daily_low = float('inf')
buying_momentum = 0
selling_momentum = 0
rolling_window 和 dow_rolling_window：存储最近的数据点用于计算移动平均。daily_high / daily_low：记录当日最高价和最低价，初始化为极端值以确保首次更新有效。buying_momentum / selling_momentum：分别累计买入和卖出动量，用于判断市场情绪。该函数用于计算当前时间距离市场开盘（09:30）经过了多少分钟。
def get_market_open_duration(window):
# 从窗口的最后一个元素提取当前时间
current_time = window.index[-1].time()
# 获取上一个交易日日期
previous_trading_day = datetime.today() - timedelta(days=1)
# 组合日期和时间
current_datetime = datetime.combine(previous_trading_day, current_time)
# 定义市场开盘时间为 09:30:00
market_start_time = datetime.combine(
previous_trading_day,
datetime.strptime("09:30:00", "%H:%M:%S").time()
)
# 计算开盘时长（分钟）
market_open_duration = (current_datetime - market_start_time).total_seconds() / 60
return market_open_duration
本系统主要关注以下技术指标：
这是系统的核心部分。我们将计算好的指标填入 Prompt，发送给 Ollama 中的 Llama 3 模型，获取专业的市场分析。
def get_natural_language_insights(
rolling_avg, ema, rsi, bollinger_upper, bollinger_lower,
price_change, volume_change, dow_rolling_avg, market_open_duration,
dow_price_change, dow_volume_change, daily_high, daily_low,
buying_momentum, selling_momentum
):
prompt = f"""
You are a professional stock broker. Apple's stock has a 5-minute rolling average of {rolling_avg:.2f}.
The Exponential Moving Average (EMA) is {ema:.2f}, and the Relative Strength Index (RSI) is {rsi:.2f}.
The Bollinger Bands are set with an upper band of {bollinger_upper:.2f} and a lower band of {bollinger_lower:.2f}.
The price has changed by {price_change:.2f}, and the volume has shifted by {volume_change}.
The DOW price has changed by {dow_price_change:.2f}, and the volume has shifted by {dow_volume_change}.
Meanwhile, the Dow Jones index has a 5-minute rolling average of {dow_rolling_avg:.2f}.
The market has been open for {market_open_duration:.2f} minutes.
Today's high was {daily_high:.2f} and low was {daily_low:.2f}.
The buying momentum is {buying_momentum:.2f} and selling momentum is {selling_momentum:.2f}.
Based on this data, provide insights into the current stock trend and the general market sentiment.
The insights should not be longer than 100 words and should not have an introduction.
"""
response = ollama.chat(
model="llama3",
messages=[{"role": "user", "content": prompt}]
)
response_text = response['message']['content'].strip()
print("Natural Language Insight:", response_text)
return response_text
此函数负责处理数据窗口，计算所有必要的技术指标。
def calculate_insights(window, dow_window):
if len(window) >= 5:
# 计算收盘价 5 分钟滚动平均值
rolling_avg = window['Close'].rolling(window=5).mean().iloc[-1]
# 计算价格和成交量变化
price_change = window['Close'].iloc[-1] - window['Close'].iloc[-2] if len(window) >= 2 else 0
volume_change = window['Volume'].iloc[-1] - window['Volume'].iloc[-2] if len(window) >= 2 else 0
# 计算道琼斯指数变化
dow_price_change = dow_window['Close'].iloc[-1] - dow_window['Close'].iloc[-2] if len(dow_window) >= 2 else 0
dow_volume_change = dow_window['Volume'].iloc[-1] - dow_window['Volume'].iloc[-2] if len(dow_window) >= 2 else 0
# 计算 EMA 和布林带
ema = window['Close'].ewm(span=5, adjust=False).mean().iloc[-1]
std = window['Close'].rolling(window=5).std().iloc[-1]
bollinger_upper = rolling_avg + (2 * std)
bollinger_lower = rolling_avg - (2 * std)
# 计算 RSI (标准周期为 14)
delta = window['Close'].diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
avg_gain = gain.rolling(window=14, min_periods=1).mean().iloc[-1]
avg_loss = loss.rolling(window=14, min_periods=1).mean().iloc[-1]
rs = avg_gain / avg_loss if avg_loss != 0 else float('nan')
rsi = 100 - (100 / (1 + rs))
# 计算道琼斯滚动平均
dow_rolling_avg = dow_window['Close'].rolling(window=5).mean().iloc[-1]
market_open_duration = get_market_open_duration(window)
# 打印计算结果
print(f"5-minute Rolling Average: {rolling_avg:.2f}")
print(f"EMA: {ema:.2f}")
print(f"RSI: {rsi:.2f}")
print(f"Bollinger Upper Band: {bollinger_upper:.2f}, Lower Band: {bollinger_lower:.2f}")
print(f"Price Change: {price_change:.2f}")
print(f"Volume Change: {volume_change}")
print(f"DOW Price Change: {dow_price_change:.2f}")
print(f"DOW Volume Change: {dow_volume_change}")
print(f"Dow Jones 5-minute Rolling Average: {dow_rolling_avg:.2f}")
print(f"Daily High: {daily_high:.2f}, Daily Low: {daily_low:.2f}")
print(f"Buying Momentum: {buying_momentum:.2f}, Selling Momentum: {selling_momentum:.2f}")
print(f"Market has been open for {market_open_duration:.2f} minutes")
# 每 5 分钟触发一次 LLM 分析
if int(market_open_duration) % 5 == 0:
insight = get_natural_language_insights(
rolling_avg, ema, rsi, bollinger_upper, bollinger_lower,
price_change, volume_change, dow_rolling_avg, market_open_duration,
dow_price_change, dow_volume_change, daily_high, daily_low,
buying_momentum, selling_momentum
)
return insight
return None
模拟每分钟接收新的数据点，并更新滚动窗口和动量指标。
def process_stock_update():
global rolling_window, data, dow_rolling_window, dow_data
global daily_high, daily_low, buying_momentum, selling_momentum
if not data.empty and not dow_data.empty:
# 模拟接收新数据点
update = data.iloc[0].to_frame().T
time_str = update.index[0].time()
print(time_str)
dow_update = dow_data.iloc[0].to_frame().T
# 移除已处理的第一行
data = data.iloc[1:]
dow_data = dow_data.iloc[1:]
# 追加到滚动窗口
rolling_window = pd.concat([rolling_window, update], ignore_index=False)
dow_rolling_window = pd.concat([dow_rolling_window, dow_update], ignore_index=False)
# 更新每日高低点
daily_high = max(daily_high, update['Close'].values[0])
daily_low = min(daily_low, update['Close'].values[0])
# 计算动量
if len(rolling_window) >= 2:
price_change = update['Close'].values[0] - rolling_window['Close'].iloc[-2]
if price_change > 0:
buying_momentum += price_change
else:
selling_momentum += abs(price_change)
# 限制滚动窗口大小为 5 分钟
if len(rolling_window) > 5:
rolling_window = rolling_window.iloc[1:]
if len(dow_rolling_window) > 5:
dow_rolling_window = dow_rolling_window.iloc[1:]
# 计算洞察
calculate_insights(rolling_window, dow_rolling_window)
使用 schedule 库模拟每分钟接收更新的场景。
schedule.every(10).seconds.do(process_stock_update)
print("Starting real-time simulation for AAPL stock updates...")
while True:
schedule.run_pending()
time.sleep(1)
为了更直观地展示分析结果，我们可以使用 Streamlit 构建一个简单的 Web 界面。以下是完整的 UI 代码示例，包含日志显示和 AI 洞察展示区域。
import streamlit as st
st.title("AI Stock Advisor")
# 创建空容器用于动态更新
log_container = st.empty()
insight_container = st.empty()
# 初始化消息
initial_msg = "Starting real-time simulation for AAPL stock updates. First update will be processed in 5 minutes..."
with st.chat_message("assistant"):
st.write(initial_msg)
# 模拟回调函数更新 UI
# 在实际应用中，建议将后台线程与 Streamlit 的 rerun 机制解耦
# 此处演示如何更新日志和洞察内容
def update_ui(last_insight=None):
log_txt = f"Last Update Time: {datetime.now().strftime('%H:%M:%S')}"
log_container.caption(log_txt)
if last_insight:
with st.chat_message("assistant"):
st.markdown(last_insight)
else:
with st.chat_message("assistant"):
st.write("Waiting for next analysis cycle...")
# 注意：Streamlit 脚本每次交互都会重新运行，因此实际部署时
# 需配合 threading 模块将定时任务放入后台线程，并通过 st.session_state 共享数据
# 以下为简化版结构示意
if __name__ == "__main__":
# 启动后台调度线程的逻辑应在此处封装
pass
本教程展示了一个完整的技术栈整合方案：
yfinance 获取金融数据。pandas 高效计算 RSI、EMA、布林带等技术指标。ollama 调用本地 Llama 3 模型，将结构化数据转化为自然语言的市场观点。Streamlit 快速搭建可视化面板。这种架构不仅适用于股票分析，也可扩展至加密货币、外汇等其他金融市场的量化监控场景。通过本地部署大模型，用户可以在保护数据隐私的前提下享受 AI 带来的分析便利。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online