某只股票盘整分析pytthon程序解析

格力电器 000651

根据你提供的信息和代码片段，这个程序是一个股票盘整分析工具，主要用于识别股票价格的盘整区间并进行技术分析。下面是对程序功能的详细解释：程序核心功能

股票数据获取：

程序使用baostock库获取股票的历史交易数据
包括开盘价、收盘价、最高价、最低价和成交量等信息

技术指标计算：

计算布林带（Bollinger Bands）指标，包括上轨、中轨和下轨
计算相对强弱指标（RSI），用于衡量股票的超买超卖状态
计算平均真实波动幅度百分比（ATR%），用于衡量价格波动性

盘整区间识别：

程序能够自动识别股票价格的盘整区间
盘整通常表现为价格在一定范围内波动，没有明显的上涨或下跌趋势
可能通过比较价格与布林带的关系、RSI值的稳定性等来判断

数据可视化：

使用matplotlib创建图表展示分析结果
包含至少两个子图：
- 第一个子图显示股票价格和布林带指标
- 第二个子图显示RSI指标和ATR百分比
程序会在图表中标记识别出的盘整区间

技术指标解释

布林带（Bollinger Bands）：

中轨（middle_band）：通常是20日移动平均线
上轨（upper_band）：中轨加上2倍标准差
下轨（lower_band）：中轨减去2倍标准差
当价格贴近上轨或下轨时，可能表示超买或超卖
当布林带收窄时，通常预示着市场即将出现大的波动

相对强弱指标（RSI）：

取值范围0-100，通常以30和70作为超卖和超买的界限
RSI>70表示超买，可能面临回调
RSI<30表示超卖，可能即将反弹
在盘整区间内，RSI通常在30-70之间波动

平均真实波动幅度百分比（ATR%）：

衡量价格波动性的指标，以百分比形式表示
低ATR%值通常出现在盘整区间，表示波动性较小
高ATR%值通常出现在趋势明显的市场，表示波动性较大

程序结构推测

基于常见的Python数据分析程序结构，该程序可能包含以下几个主要函数：

数据获取函数：负责从baostock获取股票数据
指标计算函数：计算布林带、RSI、ATR等技术指标
盘整识别函数：根据设定的条件识别盘整区间
可视化函数：创建图表展示分析结果
主函数：协调各个功能模块的执行流程

实际应用价值

交易决策辅助：通过识别盘整区间，可以帮助交易者判断突破时机
风险控制：了解股票的波动特性，有助于设置合理的止损和止盈位
趋势判断：结合多个技术指标，可以更准确地判断股票的未来走势
量化分析基础：为进一步的量化交易策略开发提供技术支持

这个程序适合对股票技术分析感兴趣的投资者和交易者使用，通过可视化的方式直观地展示股票的盘整特性和技术指标状态，帮助做出更明智的投资决策。

python源代码：

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import baostock as bs
import time
import matplotlib.font_manager as fm

# 获取股票数据
def get_stock_data(stock_code, start_date, end_date):
    # 登录baostock
    lg = bs.login()
    if lg.error_code != '0':
        print(f"登录失败: {lg.error_msg}")
        # 如果登录失败，使用模拟数据
        return generate_sample_data(start_date, end_date)
    
    # 获取日线数据
    rs = bs.query_history_k_data_plus(
        stock_code,
        "date,open,high,low,close,volume",
        start_date=start_date,
        end_date=end_date,
        frequency="d",
        adjustflag="3"  # 不复权
    )
    
    # 处理数据
    if rs.error_code != '0':
        print(f"获取数据失败: {rs.error_msg}")
        bs.logout()
        # 如果获取数据失败，使用模拟数据
        return generate_sample_data(start_date, end_date)
    
    # 转化为DataFrame
    data_list = []
    while (rs.error_code == '0') & rs.next():
        data_list.append(rs.get_row_data())
    
    if not data_list:
        print("未获取到数据，使用模拟数据")
        bs.logout()
        return generate_sample_data(start_date, end_date)
    
    df = pd.DataFrame(data_list, columns=rs.fields)
    
    # 数据类型转换
    df['date'] = pd.to_datetime(df['date'])
    df['open'] = df['open'].astype(float)
    df['high'] = df['high'].astype(float)
    df['low'] = df['low'].astype(float)
    df['close'] = df['close'].astype(float)
    df['volume'] = df['volume'].astype(float)
    
    # 设置索引并排序
    df = df.sort_values('date')
    df.set_index('date', inplace=True)
    
    # 登出baostock
    bs.logout()
    
    print(f"成功获取{stock_code}的{start_date}至{end_date}数据，共{len(df)}条")
    return df

# 生成模拟数据（当无法获取真实数据时使用）
def generate_sample_data(start_date, end_date):
    print("使用模拟数据进行演示")
    dates = pd.date_range(start=start_date, end=end_date)
    days = len(dates)
    
    # 生成一个有趋势和盘整的价格序列
    np.random.seed(42)
    base_price = 100
    trend = base_price + np.random.normal(0, 2, days).cumsum()
    
    # 在中间部分添加盘整行情
    consolidation_start = int(days * 0.3)
    consolidation_end = int(days * 0.7)
    consolidation_mean = trend[consolidation_start]
    trend[consolidation_start:consolidation_end] = consolidation_mean + np.random.normal(0, 1.5, consolidation_end-consolidation_start)
    
    df = pd.DataFrame({
        'open': trend - np.random.uniform(0, 1, days),
        'high': trend + np.random.uniform(0.5, 2, days),
        'low': trend - np.random.uniform(0.5, 2, days),
        'close': trend,
        'volume': np.random.randint(10000, 100000, days)
    }, index=dates)
    
    return df

# 手动计算布林带宽度（不使用TA-Lib）
def calculate_bollinger_band_width(df, window=20):
    # 计算移动平均线
    df['middle_band'] = df['close'].rolling(window=window).mean()
    # 计算标准差
    rolling_std = df['close'].rolling(window=window).std()
    # 计算上下布林带
    df['upper_band'] = df['middle_band'] + (rolling_std * 2)
    df['lower_band'] = df['middle_band'] - (rolling_std * 2)
    # 计算布林带宽度
    df['bb_width'] = (df['upper_band'] - df['lower_band']) / df['middle_band']
    return df

# 手动计算RSI（不使用TA-Lib）
def calculate_rsi(df, window=14):
    delta = df['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=window).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=window).mean()
    
    # 避免除零错误
    rs = gain / loss.replace(0, np.nan)
    df['rsi'] = 100 - (100 / (1 + rs))
    return df

# 手动计算ATR（不使用TA-Lib）
def calculate_atr(df, window=14):
    # 计算每日真实波动幅度
    df['tr1'] = df['high'] - df['low']
    df['tr2'] = abs(df['high'] - df['close'].shift(1))
    df['tr3'] = abs(df['low'] - df['close'].shift(1))
    df['tr'] = df[['tr1', 'tr2', 'tr3']].max(axis=1)
    
    # 计算ATR（移动平均真实波动幅度）
    df['atr'] = df['tr'].rolling(window=window).mean()
    
    # 清理临时列
    df.drop(['tr1', 'tr2', 'tr3', 'tr'], axis=1, inplace=True)
    
    # 计算ATR百分比（相对于收盘价）
    df['atr_pct'] = df['atr'] / df['close'] * 100
    return df

# 判断是否处于盘整状态
def detect_consolidation(df, lookback_period=20):
    # 计算各种技术指标
    df = calculate_bollinger_band_width(df)
    df = calculate_rsi(df)
    df = calculate_atr(df)
    
    # 移除NaN值
    df = df.dropna()
    
    if len(df) < lookback_period:
        print(f"数据不足{lookback_period}天，无法进行盘整判断")
        return False, [], {}
    
    # 计算最近一段时间的平均指标值
    recent_df = df.tail(lookback_period)
    avg_bb_width = recent_df['bb_width'].mean()
    avg_rsi = recent_df['rsi'].mean()
    avg_atr_pct = recent_df['atr_pct'].mean()
    
    # 计算价格变动范围
    price_range_pct = (recent_df['close'].max() - recent_df['close'].min()) / recent_df['close'].min() * 100
    
    # 设置盘整判断阈值
    bb_width_threshold = 0.05  # 布林带宽度阈值
    rsi_threshold = 15  # RSI波动范围阈值
    atr_pct_threshold = 1.5  # ATR百分比阈值
    price_range_threshold = 8  # 价格变动范围阈值
    
    # 判断是否处于盘整状态
    is_consolidating = False
    signals = []
    
    if avg_bb_width < bb_width_threshold:
        signals.append("布林带收窄")
    if abs(avg_rsi - 50) < rsi_threshold:
        signals.append("RSI在中间区域")
    if avg_atr_pct < atr_pct_threshold:
        signals.append("波动率低")
    if price_range_pct < price_range_threshold:
        signals.append("价格波动范围小")
    
    # 如果满足多个条件，则判断为盘整
    if len(signals) >= 3:
        is_consolidating = True
    
    return is_consolidating, signals, {
        'avg_bb_width': avg_bb_width,
        'avg_rsi': avg_rsi,
        'avg_atr_pct': avg_atr_pct,
        'price_range_pct': price_range_pct
    }

# 可视化结果
# 在导入matplotlib后添加简单的字体设置
import matplotlib.pyplot as plt
# 只保留最基本的Windows中文字体设置
plt.rcParams["font.family"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False  # 解决负号显示问题

def plot_consolidation(df, consolidation_periods, stock_code):
    plt.figure(figsize=(14, 8))
    
    # 绘制价格和布林带
    plt.subplot(2, 1, 1)
    plt.plot(df.index, df['close'], label='收盘价')
    plt.plot(df.index, df['upper_band'], 'r--', label='上布林带')
    plt.plot(df.index, df['middle_band'], 'g--', label='中布林带')
    plt.plot(df.index, df['lower_band'], 'r--', label='下布林带')
    
    # 标记盘整区间
    for period in consolidation_periods:
        if len(period) > 0:
            plt.axvspan(period[0], period[-1], color='gray', alpha=0.3)
    
    plt.title(f'{stock_code} 股票价格与布林带')
    plt.legend()
    
    # 绘制RSI和ATR百分比
    plt.subplot(2, 1, 2)
    plt.plot(df.index, df['rsi'], label='RSI')
    plt.axhline(70, color='r', linestyle='--')
    plt.axhline(30, color='g', linestyle='--')
    plt.axhline(50, color='b', linestyle='--', alpha=0.5)
    plt.ylabel('RSI')
    plt.legend(loc='upper left')
    
    plt.twinx()
    plt.plot(df.index, df['atr_pct'], 'c-', label='ATR百分比')
    plt.ylabel('ATR百分比 (%)', color='c')
    plt.grid(True, alpha=0.3)
    plt.title('RSI与ATR百分比')
    plt.legend(loc='upper right')
    
    plt.tight_layout()
    plt.show()

# 主函数
def main():
    # 获取股票数据
    #stock_code = 'sh.600938'  # 中国海油
    stock_code = 'sz.000651' # 格力电器
    start_date = '2024-01-01'
    end_date = '2025-12-31'
    
    df = get_stock_data(stock_code, start_date, end_date)
    
    # 检测最近一段时间是否处于盘整状态
    lookback_period = 20
    is_consolidating, signals, metrics = detect_consolidation(df, lookback_period)
    
    print(f"股票代码: {stock_code}")
    print(f"当前是否处于盘整状态: {'是' if is_consolidating else '否'}")
    print(f"盘整信号: {', '.join(signals) if signals else '无'}")
    if metrics:
        print(f"指标值: 布林带宽度={metrics['avg_bb_width']:.4f}, RSI={metrics['avg_rsi']:.2f}")
        print(f"        ATR百分比={metrics['avg_atr_pct']:.2f}%, 价格波动范围={metrics['price_range_pct']:.2f}%")
    
    # 检测整个时间序列中的盘整区间
    window_size = 20
    step = 5
    consolidation_periods = []
    
    # 先预处理数据以避免重复计算
    df = calculate_bollinger_band_width(df)
    df = calculate_rsi(df)
    df = calculate_atr(df)
    df = df.dropna()
    
    if len(df) >= window_size:
        for i in range(0, len(df) - window_size + 1, step):
            window_df = df.iloc[i:i+window_size][['close', 'high', 'low', 'bb_width', 'rsi', 'atr_pct']]
            
            # 计算窗口内的指标
            avg_bb_width = window_df['bb_width'].mean()
            avg_rsi = window_df['rsi'].mean()
            avg_atr_pct = window_df['atr_pct'].mean()
            price_range_pct = (window_df['close'].max() - window_df['close'].min()) / window_df['close'].min() * 100
            
            # 判断是否盘整
            signals_count = 0
            if avg_bb_width < 0.05: signals_count += 1
            if abs(avg_rsi - 50) < 15: signals_count += 1
            if avg_atr_pct < 1.5: signals_count += 1
            if price_range_pct < 8: signals_count += 1
            
            if signals_count >= 3:
                consolidation_periods.append(window_df.index)
    
    # 可视化结果
    plot_consolidation(df, consolidation_periods, stock_code)

if __name__ == "__main__":
    main()

一	二	三	四	五	六	日
		1	2	3	4	5
6	7	8	9	10	11	12
13	14	15	16	17	18	19
20	21	22	23	24	25	26
27	28	29	30	31

根据你提供的信息和代码片段，这个程序是一个股票盘整分析工具，主要用于识别股票价格的盘整区间并进行技术分析。下面是对程序功能的详细解释： 程序核心功能

根据你提供的信息和代码片段，这个程序是一个股票盘整分析工具，主要用于识别股票价格的盘整区间并进行技术分析。下面是对程序功能的详细解释：程序核心功能