当前位置：首页 > web >正文

K线涨跌概率分析系统

web 2025/7/13 5:31:45

K线涨跌概率分析系统

1. 系统概述

本系统旨在分析金融市场中连续K线模式后第一根K线的涨跌情况及其出现概率。该系统能够帮助交易者识别特定K线模式后的价格行为倾向，为交易决策提供统计依据。

2. 系统架构

2.1 数据获取模块

从数据源获取历史K线数据
数据清洗与预处理

2.2 模式识别模块

识别连续K线模式
计算模式特征

2.3 统计分析模块

统计后续K线涨跌情况
计算概率分布

2.4 可视化模块

生成统计图表
输出分析报告

3. 完整实现代码

import numpy as np
import pandas as pd
from typing import List, Dict, Tuple, Optional
import matplotlib.pyplot as plt
import seaborn as sns
from tqdm import tqdm
import warnings
warnings.filterwarnings('ignore')plt.style.use('seaborn')
plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
plt.rcParams['axes.unicode_minus'] = False  # 用来正常显示负号class KLineAnalyzer:"""K线模式分析器，用于分析连续K线模式后第一根K线的涨跌情况及其概率"""def __init__(self, data: pd.DataFrame):"""初始化分析器参数:data: 包含K线数据的DataFrame，必须包含以下列:- 'open': 开盘价- 'high': 最高价- 'low': 最低价- 'close': 收盘价- 'volume': 成交量(可选)"""required_columns = {'open', 'high', 'low', 'close'}if not required_columns.issubset(data.columns):missing = required_columns - set(data.columns)raise ValueError(f"数据缺少必要的列: {missing}")self.data = data.copy()self._preprocess_data()self.pattern_stats = {}def _preprocess_data(self):"""数据预处理"""# 确保数据按时间升序排列self.data.sort_index(inplace=True)# 计算涨跌幅self.data['pct_change'] = self.data['close'].pct_change()# 计算K线实体大小和影线长度self.data['body'] = abs(self.data['close'] - self.data['open'])self.data['upper_shadow'] = self.data['high'] - self.data[['open', 'close']].max(axis=1)self.data['lower_shadow'] = self.data[['open', 'close']].min(axis=1) - self.data['low']# 判断K线方向 (1: 阳线, -1: 阴线, 0: 平线)self.data['direction'] = np.where(self.data['close'] > self.data['open'], 1,np.where(self.data['close'] < self.data['open'], -1, 0))# 计算K线实体占比total_range = self.data['high'] - self.data['low']self.data['body_ratio'] = np.where(total_range > 0, self.data['body'] / total_range, 0)def detect_single_kline_pattern(self, row: pd.Series) -> str:"""识别单根K线形态参数:row: 单根K线的数据返回:str: K线形态名称"""body = row['body']upper = row['upper_shadow']lower = row['lower_shadow']direction = row['direction']body_ratio = row['body_ratio']if body_ratio < 0.1:return '十字星'elif body_ratio > 0.7:if direction == 1:return '大阳线' if body > body.quantile(0.75) else '阳线'else:return '大阴线' if body > body.quantile(0.75) else '阴线'elif upper > 2 * body and lower > 2 * body:return '长腿十字'elif upper > 2 * body and direction == -1:return '流星线'elif lower > 2 * body and direction == 1:return '锤子线'elif upper > lower * 2:return '上影线长'elif lower > upper * 2:return '下影线长'else:return '普通K线'def detect_consecutive_patterns(self, window: int = 3) -> pd.Series:"""识别连续K线模式参数:window: 连续K线的数量返回:pd.Series: 每window根K线组成的模式"""patterns = []directions = self.data['direction'].valuespattern_names = []for i in range(len(self.data) - window + 1):current_window = directions[i:i+window]pattern = ''.join(['U' if d == 1 else 'D' if d == -1 else 'F' for d in current_window])patterns.append(pattern)# 为常见模式命名if pattern == 'UUU':pattern_names.append('三连阳')elif pattern == 'DDD':pattern_names.append('三连阴')elif pattern == 'UDU':pattern_names.append('两阳夹一阴')elif pattern == 'DUD':pattern_names.append('两阴夹一阳')elif pattern.count('U') >= 2:pattern_names.append('多阳组合')elif pattern.count('D') >= 2:pattern_names.append('多阴组合')else:pattern_names.append('混合模式')# 将模式添加到DataFrame中pattern_series = pd.Series(patterns, index=self.data.index[window-1:])pattern_name_series = pd.Series(pattern_names, index=self.data.index[window-1:])self.data[f'{window}K_pattern'] = pattern_seriesself.data[f'{window}K_pattern_name'] = pattern_name_seriesreturn pattern_seriesdef analyze_pattern_outcomes(self, window: int = 3, look_ahead: int = 1) -> Dict:"""分析特定连续K线模式后的涨跌情况参数:window: 连续K线的数量look_ahead: 观察后续多少根K线的表现返回:Dict: 包含各模式统计结果的字典"""if f'{window}K_pattern' not in self.data.columns:self.detect_consecutive_patterns(window)pattern_col = f'{window}K_pattern'pattern_name_col = f'{window}K_pattern_name'# 准备结果字典results = {}# 获取所有唯一模式unique_patterns = self.data[pattern_col].unique()for pattern in unique_patterns:if pd.isna(pattern):continue# 获取该模式出现的所有位置pattern_indices = self.data[self.data[pattern_col] == pattern].indexif len(pattern_indices) < 5:  # 忽略出现次数过少的模式continue# 获取后续look_ahead根K线的涨跌幅outcomes = []for idx in pattern_indices:if idx + look_ahead >= len(self.data):continuenext_close = self.data.loc[idx + look_ahead, 'close']current_close = self.data.loc[idx, 'close']pct_change = (next_close - current_close) / current_closeoutcomes.append(pct_change)if not outcomes:continue# 计算统计指标outcomes = np.array(outcomes)mean_return = np.mean(outcomes)median_return = np.median(outcomes)positive_ratio = np.mean(outcomes > 0)std_dev = np.std(outcomes)# 获取模式名称pattern_name = self.data.loc[pattern_indices[0], pattern_name_col]# 存储结果results[pattern] = {'pattern_name': pattern_name,'count': len(outcomes),'mean_return': mean_return,'median_return': median_return,'positive_ratio': positive_ratio,'std_dev': std_dev,'outcomes': outcomes}# 按出现次数排序sorted_results = dict(sorted(results.items(), key=lambda x: x[1]['count'], reverse=True))self.pattern_stats[window] = sorted_resultsreturn sorted_resultsdef analyze_multiple_windows(self, windows: List[int] = [2, 3, 4]) -> Dict:"""分析多个不同窗口长度的K线模式参数:windows: 要分析的窗口长度列表返回:Dict: 包含所有窗口分析结果的字典"""all_results = {}for window in windows:print(f"正在分析 {window}K 线模式...")results = self.analyze_pattern_outcomes(window)all_results[window] = resultsreturn all_resultsdef plot_pattern_stats(self, window: int = 3, top_n: int = 15):"""绘制特定窗口长度的模式统计图参数:window: 窗口长度top_n: 显示出现频率最高的前n个模式"""if window not in self.pattern_stats:raise ValueError(f"尚未分析 {window}K 线模式，请先调用 analyze_pattern_outcomes()")stats = self.pattern_stats[window]if not stats:print(f"没有找到足够的 {window}K 线模式数据")return# 准备数据patterns = []counts = []positive_ratios = []mean_returns = []for i, (pattern, data) in enumerate(stats.items()):if i >= top_n:breakpatterns.append(f"{data['pattern_name']}\n({pattern})")counts.append(data['count'])positive_ratios.append(data['positive_ratio'])mean_returns.append(data['mean_return'])# 创建图表fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(12, 15))fig.suptitle(f'{window}K线模式统计 (前{top_n}个最常见模式)', fontsize=16)# 模式出现次数ax1.barh(patterns[::-1], counts[::-1], color='skyblue')ax1.set_title('模式出现次数')ax1.set_xlabel('出现次数')# 后续上涨概率ax2.barh(patterns[::-1], positive_ratios[::-1], color='lightgreen')ax2.set_title('后续K线上涨概率')ax2.set_xlabel('上涨概率')ax2.set_xlim(0, 1)ax2.axvline(0.5, color='red', linestyle='--', alpha=0.5)# 平均收益率ax3.barh(patterns[::-1], mean_returns[::-1], color='salmon')ax3.set_title('后续K线平均收益率')ax3.set_xlabel('平均收益率')ax3.axvline(0, color='red', linestyle='--', alpha=0.5)plt.tight_layout()plt.show()def plot_pattern_outcome_distribution(self, pattern: str, window: int = 3, bins: int = 30):"""绘制特定模式后续表现的分布图参数:pattern: 模式字符串 (如 'UUU')window: 窗口长度bins: 直方图的箱数"""if window not in self.pattern_stats:raise ValueError(f"尚未分析 {window}K 线模式，请先调用 analyze_pattern_outcomes()")stats = self.pattern_stats[window]if pattern not in stats:raise ValueError(f"未找到模式 {pattern} 的统计数据")data = stats[pattern]outcomes = data['outcomes']plt.figure(figsize=(10, 6))sns.histplot(outcomes, bins=bins, kde=True, color='blue')plt.axvline(0, color='red', linestyle='--', alpha=0.5)plt.axvline(np.mean(outcomes), color='green', linestyle='-', alpha=0.8)plt.title(f"模式 {pattern} ({data['pattern_name']}) 后续K线收益率分布\n"f"出现次数: {data['count']} | 上涨概率: {data['positive_ratio']:.2%} | "f"平均收益率: {data['mean_return']:.2%}")plt.xlabel('收益率')plt.ylabel('出现次数')plt.show()def generate_report(self, window: int = 3, min_count: int = 10) -> pd.DataFrame:"""生成分析报告参数:window: 窗口长度min_count: 模式最小出现次数阈值返回:pd.DataFrame: 包含分析结果的表格"""if window not in self.pattern_stats:raise ValueError(f"尚未分析 {window}K 线模式，请先调用 analyze_pattern_outcomes()")stats = self.pattern_stats[window]report_data = []for pattern, data in stats.items():if data['count'] < min_count:continuereport_data.append({'模式': pattern,'模式名称': data['pattern_name'],'出现次数': data['count'],'上涨概率': f"{data['positive_ratio']:.2%}",'平均收益率': f"{data['mean_return']:.2%}",'收益率标准差': f"{data['std_dev']:.2%}",'中位数收益率': f"{data['median_return']:.2%}"})df = pd.DataFrame(report_data)df.sort_values('出现次数', ascending=False, inplace=True)return df# 示例用法
if __name__ == "__main__":# 示例数据加载 (实际使用时替换为真实数据)print("正在生成示例数据...")np.random.seed(42)n = 5000dates = pd.date_range(start='2020-01-01', periods=n)base_prices = np.cumsum(np.random.normal(0, 1, n)) + 100data = pd.DataFrame({'open': base_prices,'high': base_prices + abs(np.random.normal(0.5, 0.3, n)),'low': base_prices - abs(np.random.normal(0.5, 0.3, n)),'close': base_prices + np.random.normal(0, 0.2, n)}, index=dates)# 创建分析器实例analyzer = KLineAnalyzer(data)# 分析3根连续K线模式print("\n正在分析3K线模式...")results_3k = analyzer.analyze_pattern_outcomes(window=3)# 生成报告report_3k = analyzer.generate_report(window=3)print("\n3K线模式分析报告:")print(report_3k.head(10))# 可视化analyzer.plot_pattern_stats(window=3, top_n=15)# 分析特定模式的收益率分布if 'UUU' in results_3k:analyzer.plot_pattern_outcome_distribution('UUU', window=3)# 分析多个不同窗口长度print("\n正在分析多种K线组合模式...")all_results = analyzer.analyze_multiple_windows(windows=[2, 3, 4])# 生成2K线模式报告report_2k = analyzer.generate_report(window=2)print("\n2K线模式分析报告:")print(report_2k.head(10))

4. 代码解析

4.1 数据预处理

_preprocess_data()方法对原始K线数据进行预处理，计算以下衍生指标：

涨跌幅(pct_change)
K线实体大小(body)
上下影线长度(upper_shadow, lower_shadow)
K线方向(direction): 阳线(1)、阴线(-1)、平线(0)
实体占比(body_ratio): 实体占整个K线长度的比例

4.2 单根K线模式识别

detect_single_kline_pattern()方法识别单根K线的形态，包括：

十字星
大阳线/大阴线
长腿十字
流星线
锤子线
上影线长/下影线长
普通K线

4.3 连续K线模式识别

detect_consecutive_patterns()方法识别连续window根K线组成的模式，使用U(阳线)、D(阴线)、F(平线)编码，例如：

"UUU"表示三连阳
"DUD"表示两阴夹一阳
"UDU"表示两阳夹一阴

4.4 模式结果分析

analyze_pattern_outcomes()方法统计每种连续K线模式后look_ahead根K线的表现，计算：

出现次数
平均收益率
中位数收益率
上涨概率
收益率标准差

4.5 可视化分析

系统提供多种可视化方法：

plot_pattern_stats(): 展示最常见模式的统计指标
plot_pattern_outcome_distribution(): 展示特定模式后续收益率的分布
generate_report(): 生成详细的统计分析表格

5. 统计分析方法

5.1 概率计算原理

系统采用频率学派方法计算上涨概率：

上涨概率 = (后续上涨的次数) / (模式出现的总次数)

5.2 统计显著性检验

虽然完整代码中没有直接实现，但可以扩展添加统计检验功能，例如：

from scipy import statsdef test_significance(pattern1: str, pattern2: str, window: int = 3) -> float:"""检验两个模式后续表现差异的显著性参数:pattern1: 第一个模式pattern2: 第二个模式window: 窗口长度返回:float: p值"""if window not in self.pattern_stats:raise ValueError(f"尚未分析 {window}K 线模式")stats = self.pattern_stats[window]if pattern1 not in stats or pattern2 not in stats:raise ValueError("未找到指定的模式")outcomes1 = stats[pattern1]['outcomes']outcomes2 = stats[pattern2]['outcomes']t_stat, p_value = stats.ttest_ind(outcomes1, outcomes2)return p_value