如何用Python计算不同时间周期下股票的200 EMA并分析

一、数据获取疑问解答 #

• 对于12小时这类非标准时间周期，yfinance没有直接提供对应间隔的历史数据接口，因此需要通过1小时级别的原始数据进行聚合处理来生成12小时K线（取每个12小时窗口的收盘价），再计算EMA。
• 对于1小时、4小时、1天这类标准周期，可以直接通过yfinance的interval参数获取对应数据，也可以用更细粒度数据聚合，直接获取效率更高。

二、修改后的完整代码 #

import yfinance as yf
import pandas as pd

# 核心配置参数
tickers = "AAPL MSFT GOOGL AMZN TSLA"
target_periods = ['1h', '4h', '12h', '1d']  # 需要分析的时间周期
base_interval = '1h'  # 基础数据间隔，用于聚合非标准周期
data_period = "6mo"  # 拉长数据周期，确保能覆盖200期EMA计算需求

# 获取1小时级别基础收盘价数据
data = yf.download(
    tickers=tickers,
    period=data_period,
    interval=base_interval,
    auto_adjust=True,
    prepost=False,
    threads=True
)
close_prices = data['Close'][tickers.split()]

# 获取股票基本信息（名称、市值）
def fetch_stock_info(tickers_list):
    stock_info = {}
    for ticker in tickers_list:
        stock = yf.Ticker(ticker)
        stock_info[ticker] = {
            'Name': stock.info.get('longName', 'N/A'),
            'MarketCap': stock.info.get('marketCap', 'N/A')
        }
    return stock_info

stock_info = fetch_stock_info(tickers.split())

# 将基础数据转换为目标周期的收盘价数据
def get_period_close_data(df, target_interval):
    if target_interval == '1h':
        return df
    elif target_interval == '4h':
        return df.resample('4H').last()
    elif target_interval == '12h':
        return df.resample('12H').last()
    elif target_interval == '1d':
        return df.resample('D').last()
    else:
        raise ValueError(f"不支持的周期类型：{target_interval}")

# 计算各周期200 EMA并生成分析结果
analysis_results = {}
for ticker in tickers.split():
    ticker_results = {}
    for period in target_periods:
        # 获取对应周期的收盘价序列
        period_close = get_period_close_data(close_prices[ticker], period)
        # 计算200 EMA，dropna过滤无效数据
        period_close['200_EMA'] = period_close.ewm(span=200, adjust=False).mean().dropna()
        
        # 提取最新数据并判断股价与EMA的关系
        latest_price = period_close.iloc[-1]
        if len(period_close['200_EMA']) < 1:
            ticker_results[period] = {
                '200_EMA': '数据不足无法计算',
                'Current_Price': round(latest_price, 2),
                'Is_Above_EMA': 'N/A'
            }
            continue
        
        ema_val = period_close['200_EMA'].iloc[-1]
        is_above = latest_price > ema_val
        ticker_results[period] = {
            '200_EMA': round(ema_val, 2),
            'Current_Price': round(latest_price, 2),
            'Is_Above_EMA': '是' if is_above else '否'
        }
    
    analysis_results[ticker] = {
        'Basic_Info': stock_info[ticker],
        'Period_Analysis': ticker_results
    }

# 格式化输出分析结果
for ticker, result in analysis_results.items():
    print(f"=== {ticker} ({result['Basic_Info']['Name']}) ===")
    print(f"市值: {result['Basic_Info']['MarketCap']:,}")
    for period, details in result['Period_Analysis'].items():
        print(f"\n周期: {period}")
        print(f"200 EMA: {details['200_EMA']}")
        print(f"当前股价: {details['Current_Price']}")
        print(f"股价高于200 EMA: {details['Is_Above_EMA']}")
    print("\n" + "-"*50 + "\n")

三、关键代码说明 #

• 数据周期保障：设置data_period为6个月，确保即使是12小时周期，也能收集到至少200期有效数据（200*12h=2400小时≈100天），满足200 EMA的计算条件。
• 周期聚合逻辑：利用pandas的resample方法将1小时数据聚合为目标周期，取每个时间窗口的最后一个收盘价作为该周期的收盘价，符合传统K线收盘价定义。
• EMA计算规则：使用ewm(span=200, adjust=False)计算指数移动平均线，adjust=False采用递归式EMA算法，与主流金融分析工具的计算逻辑一致。
• 异常处理：针对数据不足无法计算EMA的情况，添加了友好提示，避免程序报错中断。

内容的提问来源于stack exchange，提问作者Ezio