回测 回撤 hline

import pandas as pd

def first(df):
    #流通市值 30 45 60 90 120 日均价 以及在4日内高点发生回撤时候的比率
    df['lt']=df['amount']/(1000000*df['turn'])
    df['hc30']=(df['high']-df['p30'])/df['p30']
    df['hc45']=(df['high']-df['p45'])/df['p45']
    df['hc60']=(df['high']-df['p60'])/df['p60']
    df['hc90']=(df['high']-df['p90'])/df['p90']
    df['hc120']=(df['high']-df['p120'])/df['p120']
    is_pivot_high = df['high'] == df['high'].rolling(window=5, center=True).max()
    high_df = df[is_pivot_high][['mcode','date','open','close','low','high','p30','p45','p60','p90','p120','amount','turn','lt','hc30','hc45','hc60','hc90','hc120']]
    return high_df

        
results=[]
# 执行每个
results.append(first(df))

final_df=pd.concat(results,ignore_index=True)
final_df.to_csv("xxx.csv",index=False,encoding="utf-8-sig")

ADD

import pymysql
from pymysql.err import OperationalError, ProgrammingError
import pandas as pd
import time



def first(df):
    #流通市值 30 45 60 90 120 日均价 以及在4日内高点发生回撤时候的比率
    df['lt']=df['amount']/(1000000*df['turn'])
    df['hc30']=(df['high']-df['p30'])/df['p30']
    df['hc45']=(df['high']-df['p45'])/df['p45']
    df['hc60']=(df['high']-df['p60'])/df['p60']
    df['hc90']=(df['high']-df['p90'])/df['p90']
    df['hc120']=(df['high']-df['p120'])/df['p120']
    is_pivot_high = df['high'] == df['high'].rolling(window=5, center=True).max()
    high_df = df[is_pivot_high][['mcode','date','open','close','low','high','p30','p45','p60','p90','p120','amount','turn','lt','hc30','hc45','hc60','hc90','hc120']]
    return high_df

def get_all_data(mcode):
    start = time.time()
    sql = "SELECT * FROM cmf_quant5 WHERE mcode = %s AND date>2025-10-01 ORDER BY date ASC"
    df= pd.read_sql(sql, conn, params=(mcode,))
     # 清理异常值（必须保留）
    df = df.replace([float('inf'), -float('inf')], 0)
    df = df.fillna(0)
    return df


# 数据库配置
config = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "password": "123456",
    "database": "gupiao",
    "charset": "utf8mb4"
}

conn = pymysql.connect(**config)
cursor = conn.cursor()

results=[]
df = pd.read_csv('/home/may/gupiao/all.csv', dtype={'code': str})
i=0
for index, row in df.iterrows():
    df=get_all_data(row['code'])
    results.append(first(df))
final_df=pd.concat(results,ignore_index=True)
final_df.to_csv("xxx.csv",index=False,encoding="utf-8-sig")