2.py

import pandas as pd
from datetime import datetime
from reportlab.lib.pagesizes import A4
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph
from reportlab.lib import colors
from reportlab.lib.enums import TA_CENTER
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
import os
import platform


# ==============================================
# 0. 准备工作：更强壮的字体注册逻辑
# ==============================================
def register_chinese_font():
    """尝试自动寻找黑体字体，兼容 Windows/Mac/Linux"""
    font_name = 'Helvetica'
    font_paths = [
        'C:/Windows/Fonts/simhei.ttf',
        'C:/Windows/Fonts/msyh.ttc',
        '/System/Library/Fonts/PingFang.ttc',
        '/usr/share/fonts/truetype/droid/DroidSansFallbackFull.ttf',
        './simhei.ttf'
    ]
    registered = False
    for path in font_paths:
        if os.path.exists(path):
            try:
                pdfmetrics.registerFont(TTFont('MyChineseFont', path))
                font_name = 'MyChineseFont'
                print(f"成功加载字体: {path}")
                registered = True
                break
            except Exception:
                continue
    if not registered:
        print("【警告】未找到支持中文的字体文件！PDF中的中文将显示为方框。")
    return font_name


font_name = register_chinese_font()

# ==============================================
# 1. 读取数据
# ==============================================
filename = "ashare_history_data.csv"
print(f"正在读取 {filename} ...")

if not os.path.exists(filename):
    print(f"错误：找不到文件 {filename}")
    exit()

df = pd.read_csv(filename)

# 预处理
df['日期'] = pd.to_datetime(df['日期'])
# 确保包含了 '成交额' 列
numeric_cols = ['开盘', '最高', '最低', '收盘', '涨跌幅', '成交额']
for col in numeric_cols:
    if col in df.columns:
        df[col] = pd.to_numeric(df[col], errors='coerce')

df = df.sort_values(by=['代码', '日期'])

if '名称' in df.columns:
    df.rename(columns={'名称': 'code_name'}, inplace=True)
if 'code_name' not in df.columns:
    df['code_name'] = '未知'
df['code_name'] = df['code_name'].fillna('未知').astype(str)

print("数据预处理完成，开始筛选...")


# ==============================================
# 2. 筛选核心逻辑 (回溯最早缺口 + 放量检测)
# ==============================================
def check_gap_strategy(group):
    # 重置索引
    group = group.sort_values('日期').reset_index(drop=True)

    if len(group) < 30: return None  # 需要足够的数据计算均线和成交量

    # 获取最新数据
    latest_row = group.iloc[-1]
    latest_close = latest_row['收盘']
    latest_date = latest_row['日期']
    latest_pct_chg = latest_row['涨跌幅']
    stock_name = latest_row['code_name']

    # 过滤 ST
    if 'ST' in stock_name.upper():
        return None

    # 1. 价格初筛 (< 20 元)
    if latest_close >= 20:
        return None

    # 2. 寻找所有“向上缺口”
    group['prev_high'] = group['最高'].shift(1)
    gap_indices = group[group['最低'] > group['prev_high']].index.tolist()

    if not gap_indices: return None

    # 3. 倒序遍历缺口（从最近的日期往回找）
    for gap_idx in gap_indices[::-1]:

        # --- [A] 回溯寻找最早的缺口 ---
        target_gap_idx = gap_idx
        current_trace_idx = gap_idx

        while current_trace_idx > 1:
            prev_idx = current_trace_idx - 1
            prev_row = group.iloc[prev_idx]

            if prev_row['涨跌幅'] > 0:
                prev_high_of_prev = group.iloc[prev_idx - 1]['最高']
                if prev_row['最低'] > prev_high_of_prev:
                    target_gap_idx = prev_idx
                current_trace_idx = prev_idx
            else:
                break

        # 此时 target_gap_idx 是我们要分析的核心缺口位置

        # ==========================================================
        # --- [B] 【新增】成交额放量检测 (> 2.0 倍) ---
        # ==========================================================
        # 确保缺口前有足够数据计算均量
        pre_gap_data = group.iloc[:target_gap_idx]
        if len(pre_gap_data) < 5: continue

        # 计算缺口前20日的平均成交额 (如果不足20日则取实际天数)
        avg_amount_pre_20 = pre_gap_data['成交额'].tail(20).mean()
        if avg_amount_pre_20 == 0: continue

        # 计算缺口后5日(含缺口当天)的平均成交额
        post_gap_window = group.iloc[target_gap_idx: target_gap_idx + 5]
        avg_amount_post_5 = post_gap_window['成交额'].mean()

        volume_ratio = avg_amount_post_5 / avg_amount_pre_20

        # 如果放量不足 2 倍，则该缺口无效，跳过
        if volume_ratio < 2.0:
            continue
        # ==========================================================

        row = group.iloc[target_gap_idx]
        gap_date = row['日期']
        gap_support_price = row['最低']
        gap_open_price = row['开盘']

        # 如果回溯后的缺口日期就是最新交易日，则忽略
        if gap_date == latest_date: continue

        # --- [C] 判断缺口是否被回补 ---
        post_gap_data = group[group['日期'] > gap_date]
        if post_gap_data.empty: continue

        min_price_after = post_gap_data['最低'].min()
        if min_price_after < gap_support_price:
            continue

        # --- [D] 判断是否“曾经阔过” (最大涨幅 > 10%) ---
        max_price_after = post_gap_data['最高'].max()
        max_increase = (max_price_after - gap_open_price) / gap_open_price

        if max_increase < 0.10:
            continue

        # --- [E] 判断是否回调到位 (距离缺口 < 15%) ---
        current_distance = (latest_close - gap_support_price) / gap_support_price

        if 0 <= current_distance <= 0.15:
            days_ago = (latest_date - gap_date).days

            return {
                '代码': row['代码'],
                '名称': stock_name,
                '缺口日期': gap_date.strftime('%Y-%m-%d'),
                '天数': days_ago,
                '缺口价': f"{gap_support_price:.2f}",
                '现价': f"{latest_close:.2f}",
                '今日涨幅': latest_pct_chg,
                '最大涨幅': f"{max_increase * 100:.1f}%",
                '离缺口': f"{current_distance * 100:.1f}%"
            }

    return None


# ==============================================
# 3. 执行筛选 (输出分桶：1/2/3个月)
# ==============================================
results_1m = []
results_2m = []
results_3m = []

grouped = df.groupby('代码')
total = len(grouped)
count = 0

for name, group in grouped:
    res = check_gap_strategy(group)
    if res:
        days = res['天数']
        if days <= 30:
            results_1m.append(res)
        elif 30 < days <= 60:
            results_2m.append(res)
        elif 60 < days <= 90:
            results_3m.append(res)

    count += 1
    if count % 1000 == 0:
        print(f"已扫描 {count}/{total}...")

print(f"\n筛选完成 (回溯最早缺口 + 放量>2倍 + 缺口未补 + 曾涨>10% + 回调到位)：")
print(f"1个月内: {len(results_1m)}")
print(f"2个月内: {len(results_2m)}")
print(f"3个月内: {len(results_3m)}")


# ==============================================
# 4. PDF 生成函数
# ==============================================
def create_beautiful_pdf(data_list, filename, main_title):
    if not data_list:
        print(f"跳过生成 {filename} (无数据)")
        return

    df_res = pd.DataFrame(data_list)

    # 颜色定义
    color_up = colors.Color(0.8, 0, 0)
    color_down = colors.Color(0, 0.5, 0)
    color_header_bg = colors.Color(0.2, 0.3, 0.5)
    color_row_even = colors.Color(0.95, 0.95, 0.95)

    # 列映射
    cols_map = {
        '代码': '代码', '名称': '名称', '缺口日期': '缺口日期',
        '天数': '天数', '缺口价': '缺口价', '现价': '现价',
        '今日涨幅': '今日%', '最大涨幅': '最大涨%', '离缺口': '离缺口%'
    }

    # 准备表头
    display_headers = list(cols_map.values())
    table_data = [display_headers]

    style_cmds = [
        ('FONTNAME', (0, 0), (-1, -1), font_name),
        ('FONTSIZE', (0, 0), (-1, -1), 10),
        ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
        ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'),
        ('BACKGROUND', (0, 0), (-1, 0), color_header_bg),
        ('TEXTCOLOR', (0, 0), (-1, 0), colors.white),
        ('FONTSIZE', (0, 0), (-1, 0), 11),
        ('BOTTOMPADDING', (0, 0), (-1, 0), 8),
        ('TOPPADDING', (0, 0), (-1, 0), 8),
        ('LINEBELOW', (0, 0), (-1, 0), 1, colors.white),
        ('LINEBELOW', (0, 1), (-1, -1), 0.5, colors.lightgrey),
    ]

    for idx, row in df_res.iterrows():
        row_idx = idx + 1

        # 获取数值型涨跌幅用于判断颜色
        try:
            raw_pct = float(row['今日涨幅'])
        except:
            raw_pct = 0

        pct_str = f"{raw_pct:.2f}%"

        row_values = [
            row['代码'], row['名称'], row['缺口日期'], row['天数'],
            row['缺口价'], row['现价'], pct_str,
            row['最大涨幅'], row['离缺口']
        ]
        table_data.append(row_values)

        if row_idx % 2 == 0:
            style_cmds.append(('BACKGROUND', (0, row_idx), (-1, row_idx), color_row_even))

        if raw_pct > 0:
            style_cmds.append(('TEXTCOLOR', (0, row_idx), (-1, row_idx), color_up))
        elif raw_pct < 0:
            style_cmds.append(('TEXTCOLOR', (0, row_idx), (-1, row_idx), color_down))

    doc = SimpleDocTemplate(filename, pagesize=A4,
                            rightMargin=30, leftMargin=30,
                            topMargin=30, bottomMargin=30)
    elements = []
    styles = getSampleStyleSheet()

    title_style = ParagraphStyle(
        name='MainTitle', parent=styles['Title'],
        fontName=font_name, fontSize=18, leading=22,
        alignment=TA_CENTER, spaceAfter=10, textColor=colors.Color(0.2, 0.2, 0.2)
    )

    subtitle_style = ParagraphStyle(
        name='SubTitle', parent=styles['Normal'],
        fontName=font_name, fontSize=10, alignment=TA_CENTER,
        spaceAfter=20, textColor=colors.grey
    )

    current_time = datetime.now().strftime("%Y-%m-%d")
    elements.append(Paragraph(main_title, title_style))
    elements.append(Paragraph(f"生成日期: {current_time} | 策略: 最早启动缺口 + 放量(>2倍) + 回调确认", subtitle_style))

    col_widths = [60, 70, 70, 35, 45, 45, 55, 55, 55]
    t = Table(table_data, colWidths=col_widths, repeatRows=1)
    t.setStyle(TableStyle(style_cmds))
    elements.append(t)

    try:
        doc.build(elements)
        print(f"成功生成: {filename}")
    except Exception as e:
        print(f"生成失败 {filename}: {e}")


# ==============================================
# 5. 生成报告
# ==============================================
print("-" * 30)
create_beautiful_pdf(results_1m, "report_clean_1month.pdf", "优质缺口策略 (近1个月)")
create_beautiful_pdf(results_2m, "report_clean_2month.pdf", "优质缺口策略 (近2个月)")
create_beautiful_pdf(results_3m, "report_clean_3month.pdf", "优质缺口策略 (近3个月)")
print("-" * 30)