基于 Ollama 与 Python 的股票技术指标分析及可视化

辅助函数实现

1. 计算市场开盘时长

该函数用于计算当前时间距离市场开盘（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

2. 技术指标详解

本系统主要关注以下技术指标：

• 滚动平均值 (Rolling Average)：反映短期价格趋势。
• 指数移动平均线 (EMA)：对近期价格赋予更高权重，比简单移动平均更敏感。
• 相对强弱指数 (RSI)：衡量价格变动的速度和变化，通常用于判断超买或超卖状态。
• 布林带 (Bollinger Bands)：由中轨（移动平均）、上轨和下轨组成，用于衡量波动性。

3. 生成自然语言洞察

这是系统的核心部分。我们将计算好的指标填入 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

4. 计算指标逻辑

此函数负责处理数据窗口，计算所有必要的技术指标。

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

5. 处理股票更新

模拟每分钟接收新的数据点，并更新滚动窗口和动量指标。

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)

6. 调度任务

使用 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 可视化界面

为了更直观地展示分析结果，我们可以使用 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

总结

本教程展示了一个完整的技术栈整合方案：

1. 数据层：利用 yfinance 获取金融数据。
2. 计算层：使用 pandas 高效计算 RSI、EMA、布林带等技术指标。
3. 智能层：通过 ollama 调用本地 Llama 3 模型，将结构化数据转化为自然语言的市场观点。
4. 展示层：使用 Streamlit 快速搭建可视化面板。

这种架构不仅适用于股票分析，也可扩展至加密货币、外汇等其他金融市场的量化监控场景。通过本地部署大模型，用户可以在保护数据隐私的前提下享受 AI 带来的分析便利。