stock/Chan241206/kline_processor.py

import pandas as pd

class KlineProcessor:
    def __init__(self, df):
        """
        初始化，处理历史数据中的包含关系
        :param df: 历史K线数据，包含 'High' 和 'Low' 列
        """
        self.df = self.process_inclusion_globally(df)
    
    def process_inclusion_globally(self, df):
        """
        全局处理包含关系，结合更早的 K 线进行方向判断
        :param df: 包含 'High' 和 'Low' 列的 DataFrame
        :return: 处理完成的无包含关系的 DataFrame
        """
        i = 1
        while i < len(df):
            # 检查 i 和 i-1 是否有包含关系
            if (df.loc[i, 'High'] <= df.loc[i-1, 'High'] and df.loc[i, 'Low'] >= df.loc[i-1, 'Low']) or \
            (df.loc[i, 'High'] >= df.loc[i-1, 'High'] and df.loc[i, 'Low'] <= df.loc[i-1, 'Low']):

                # 如果有包含关系，判断方向并合并
                if i - 2 >= 0:  # 检查是否存在 i-2
                    # 如果 i-2 存在，基于 i-2 和 i-1 的关系判断方向
                    if df.loc[i-1, 'High'] <= df.loc[i-2, 'High']:
                        # 当前为 LL 合并
                        df.loc[i-1, 'High'] = min(df.loc[i, 'High'], df.loc[i-1, 'High'])
                        df.loc[i-1, 'Low'] = min(df.loc[i, 'Low'], df.loc[i-1, 'Low'])
                    else:
                        # 当前为 HH 合并
                        df.loc[i-1, 'High'] = max(df.loc[i, 'High'], df.loc[i-1, 'High'])
                        df.loc[i-1, 'Low'] = max(df.loc[i, 'Low'], df.loc[i-1, 'Low'])
                else:
                    # 如果 i-2 不存在，仅根据 i 和 i-1 判断方向
                    if df.loc[i, 'High'] <= df.loc[i-1, 'High']:
                        # 当前为 LL 合并
                        df.loc[i-1, 'High'] = min(df.loc[i, 'High'], df.loc[i-1, 'High'])
                        df.loc[i-1, 'Low'] = min(df.loc[i, 'Low'], df.loc[i-1, 'Low'])
                    else:
                        # 当前为 HH 合并
                        df.loc[i-1, 'High'] = max(df.loc[i, 'High'], df.loc[i-1, 'High'])
                        df.loc[i-1, 'Low'] = max(df.loc[i, 'Low'], df.loc[i-1, 'Low'])

                # 删除当前 K 线
                df = df.drop(index=i).reset_index(drop=True)

                # 回溯到前一根，重新检查合并后的关系
                i = max(i - 1, 1)
            else:
                # 如果没有包含关系，检查下一根 K 线
                i += 1
        print(df.head())
        return df