test/debug_segments_analysis.py at main · pigbiglong/test · GitHub

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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
import pandas as pd
import akshare as ak
from chan_analyzer.engine import ChanEngine

# 获取更长时间段的数据
print("获取数据...")
df = ak.stock_zh_a_hist(symbol='600519', period='daily', start_date='20200101', end_date='20241231', adjust='qfq')
print(f"获取到 {len(df)} 条数据")

# 数据预处理
df.rename(columns={'日期': 'Date', '开盘': 'Open', '最高': 'High', '最低': 'Low', '收盘': 'Close', '成交量': 'Volume'}, inplace=True)
df['Date'] = pd.to_datetime(df['Date'])
df.set_index('Date', inplace=True)
df.sort_index(ascending=True, inplace=True)

# 直接分析线段生成过程
from chan_analyzer.types import Kline
from chan_analyzer.preprocessor import handle_inclusion
from chan_analyzer.fractal import find_fractals
from chan_analyzer.stroke import generate_strokes
from chan_analyzer.segment import generate_segments

print("开始逐步分析...")

# 1. 预处理K线
original_klines = []
for i, row in enumerate(df.itertuples()):
    kline = Kline(
        timestamp=int(row.Index.timestamp()),
        open=row.Open, high=row.High,
        low=row.Low, close=row.Close,
        volume=row.Volume
    )
    kline.original_index = i  # 手动添加original_index属性
    original_klines.append(kline)

processed_klines = handle_inclusion(original_klines)
print(f"预处理后K线数量: {len(processed_klines)}")

# 2. 寻找分型
fractals = find_fractals(processed_klines)
print(f"分型数量: {len(fractals)}")

# 3. 生成笔
strokes = generate_strokes(fractals)
print(f"笔数量: {len(strokes)}")

# 4. 生成线段
segments = generate_segments(strokes)
print(f"线段数量: {len(segments)}")

# 详细分析线段
if segments:
    print("\n线段详情:")
    for i, segment in enumerate(segments):
        start_time = pd.to_datetime(segment.start_stroke.start_fractal.kline.timestamp, unit='s').strftime('%Y-%m-%d')
        end_time = pd.to_datetime(segment.end_stroke.end_fractal.kline.timestamp, unit='s').strftime('%Y-%m-%d')
        print(f"  线段 {i+1}: {start_time} 到 {end_time}, 方向: {segment.direction}, 价格: {segment.low:.2f} - {segment.high:.2f}")

        # 显示构成此线段的笔
        print(f"    构成笔数: {len(segment.strokes)}")
        for j, stroke in enumerate(segment.strokes[:3]):  # 只显示前3笔
            stroke_start = pd.to_datetime(stroke.start_fractal.kline.timestamp, unit='s').strftime('%Y-%m-%d')
            stroke_end = pd.to_datetime(stroke.end_fractal.kline.timestamp, unit='s').strftime('%Y-%m-%d')
            print(f"      笔 {j+1}: {stroke_start} 到 {stroke_end}, 方向: {stroke.direction}, 价格: {stroke.low:.2f} - {stroke.high:.2f}")
        if len(segment.strokes) > 3:
            print(f"      ... 还有 {len(segment.strokes) - 3} 笔")

# 分析为什么线段数量这么少
print(f"\n总笔数: {len(strokes)}")
print("检查线段生成条件...")

# 检查是否有足够的连续三笔来形成中枢
if len(strokes) >= 3:
    print("\n检查所有连续三笔组合:")
    from chan_analyzer.pivot import has_overlap, calculate_pivot_range

    overlapping_count = 0
    for i in range(len(strokes) - 2):
        s1, s2, s3 = strokes[i], strokes[i+1], strokes[i+2]
        if has_overlap([s1, s2, s3]):
            zg, zd, gg, dd = calculate_pivot_range([s1, s2, s3])
            overlapping_count += 1
            if overlapping_count <= 5:  # 只显示前5个
                print(f"  组合 {i}-{i+2}: 有重叠, ZG={zg:.2f}, ZD={zd:.2f}")
        else:
            if i < 5:  # 只显示前5个无重叠的例子
                print(f"  组合 {i}-{i+2}: 无重叠")

    print(f"总共找到 {overlapping_count} 个有重叠的三笔组合")

print("\n分析完成")