💻 核心代码 2：每日自动增量更新引擎

魔改完成后，你只需要把它部署在电脑上，每天收盘后自动运行。代码会智能判断数据库中最后的更新日期，老股票只取增量，新上市股票自动在数据库中 ALTER 增加列并全量回测，连通达信自选板块（.blk）和 ETF 行情都能一并搞定！

import osimport structimport pymysqlimport pandas as pdfrom mootdx.reader import Readerfrom tqdm import tqdmfrom datetime import datetime, timedelta# ==================== 配置 ====================TDX_DIR = 'D:/交易/通达信'BLOCK_FILE = r'D:\交易\通达信\T0002\blocknew\QABJ.blk'STOCKS_PER_TABLE = 140# ==================== 数据库连接 ====================defconnect_to_database():try:        conn = pymysql.connect(            host='localhost',            port=3306,            user='local',  # 替换为你的数据库账号            password='123456', # 替换为你的数据库密码            db='ggsj',            charset='utf8mb4'        )return connexcept Exception as e:        print(f"数据库连接失败: {str(e)}")raise# ==================== 解析通达信 .day 文件 ====================defparse_tdx_day_file(filepath):"""解析通达信 .day 文件，返回 DataFrame"""    data = []try:with open(filepath, 'rb') as f:whileTrue:                stock_date = f.read(4)                stock_open = f.read(4)                stock_high = f.read(4)                stock_low = f.read(4)                stock_close = f.read(4)                stock_amount = f.read(4)                stock_vol = f.read(4)                stock_reservation = f.read(4)ifnot stock_date:break                stock_date = struct.unpack("l", stock_date)[0]     # 4 字节 如 20091229                stock_open = struct.unpack("l", stock_open)[0] / 100# 开盘价                stock_high = struct.unpack("l", stock_high)[0] / 100# 最高价                stock_low = struct.unpack("l", stock_low)[0] / 100# 最低价                stock_close = struct.unpack("l", stock_close)[0] / 100# 收盘价                stock_amount = struct.unpack("f", stock_amount)[0]      # 成交额                stock_vol = struct.unpack("l", stock_vol)[0] / 100# 成交量（单位：手）# 格式化日期                date_obj = datetime.strptime(str(stock_date), '%Y%m%d')                data.append({'date': date_obj,'open': stock_open,'high': stock_high,'low': stock_low,'close': stock_close,'amount': stock_amount,'volume': stock_vol                })if data:            df = pd.DataFrame(data)            df.set_index('date', inplace=True)return dfelse:returnNoneexcept Exception:returnNonedefresolve_day_file_path(code):    code = str(code).zfill(6)# 根据股票代码前缀智能判断市场优先级# 6开头 - 沪市股票，优先匹配 sh# 0、3开头 - 深市股票，优先匹配 sz  # 4、8开头 - 北交所股票，优先匹配 bjif code.startswith('6'):        candidates = [            os.path.join(TDX_DIR, 'vipdoc', 'sh', 'lday', f'sh{code}.day'),            os.path.join(TDX_DIR, 'vipdoc', 'sz', 'lday', f'sz{code}.day'),            os.path.join(TDX_DIR, 'vipdoc', 'bj', 'lday', f'bj{code}.day')        ]elif code.startswith(('0', '3')):        candidates = [            os.path.join(TDX_DIR, 'vipdoc', 'sz', 'lday', f'sz{code}.day'),            os.path.join(TDX_DIR, 'vipdoc', 'sh', 'lday', f'sh{code}.day'),            os.path.join(TDX_DIR, 'vipdoc', 'bj', 'lday', f'bj{code}.day')        ]elif code.startswith(('4', '8')):        candidates = [            os.path.join(TDX_DIR, 'vipdoc', 'bj', 'lday', f'bj{code}.day'),            os.path.join(TDX_DIR, 'vipdoc', 'sz', 'lday', f'sz{code}.day'),            os.path.join(TDX_DIR, 'vipdoc', 'sh', 'lday', f'sh{code}.day')        ]else:# 其他情况使用默认顺序        candidates = [            os.path.join(TDX_DIR, 'vipdoc', 'bj', 'lday', f'bj{code}.day'),            os.path.join(TDX_DIR, 'vipdoc', 'sh', 'lday', f'sh{code}.day'),            os.path.join(TDX_DIR, 'vipdoc', 'sz', 'lday', f'sz{code}.day')        ]for filepath in candidates:if os.path.exists(filepath):return filepathreturnNone# ==================== 解析板块文件 ====================defparse_block_file(filepath):"""解析通达信自定义板块文件，返回所有股票代码列表（包含北交所）"""    stocks = []ifnot os.path.exists(filepath):        print(f"文件不存在: {filepath}")return stockstry:with open(filepath, 'r', encoding='gbk', errors='ignore') as f:            content = f.read()            lines = content.splitlines()for line in lines:                line = line.strip()if len(line) >= 7:                    symbol = line[1:7]if symbol.isdigit() and len(symbol) == 6:                        symbol = symbol.zfill(6)                        stocks.append(symbol)except Exception as e:        print(f"解析板块文件出错: {e}")return stocks# ==================== 获取数据库结构信息 ====================defget_db_structure(conn, table_prefix):"""获取数据库中现有的表结构和包含的股票"""    cursor = None    table_mapping = {} # table_name -> [stock_codes]    all_existing_stocks = set()try:        cursor = conn.cursor()# 查找所有批次表        cursor.execute(f"SHOW TABLES LIKE '{table_prefix}%'")        tables = sorted([row[0] for row in cursor.fetchall()], key=lambda x: int(x.split('_')[-1]))for table_name in tables:# 获取列名            cursor.execute(f"SHOW COLUMNS FROM `{table_name}`")            columns = cursor.fetchall()# 提取股票代码            stocks = []for col in columns:                col_name = col[0]if'_'in col_name and col_name.endswith('_open'):                    stock_code = col_name.replace('_open', '')                    stocks.append(stock_code)                    all_existing_stocks.add(stock_code)            table_mapping[table_name] = stocksreturn table_mapping, all_existing_stocks, tablesfinally:if cursor:            cursor.close()# ==================== 更新数据库结构 ====================defupdate_schema(conn, all_stocks_from_file, table_prefix):"""更新数据库结构：添加新股票列或创建新表"""    cursor = Nonetry:        cursor = conn.cursor()# 获取当前数据库结构        table_mapping, existing_stocks, tables = get_db_structure(conn, table_prefix)# 找出新增的股票        new_stocks = [s for s in all_stocks_from_file if s notin existing_stocks]ifnot new_stocks:            print("没有发现新上市的代码")return table_mapping, []        print(f"发现 {len(new_stocks)} 只新代码，正在更新数据库结构...")# 处理新股票        current_new_stock_idx = 0# 1. 检查最后一个表是否已满if tables:            last_table = tables[-1]            current_count = len(table_mapping[last_table])if current_count < STOCKS_PER_TABLE:# 还可以往最后一个表添加                slots_available = STOCKS_PER_TABLE - current_count                stocks_to_add = new_stocks[current_new_stock_idx : current_new_stock_idx + slots_available]if stocks_to_add:                    print(f"  向 {last_table} 添加 {len(stocks_to_add)} 只新代码...")# 构建 ALTER TABLE 语句                    alter_parts = []for code in stocks_to_add:                        alter_parts.append(f"ADD COLUMN `{code}_open` DECIMAL(10,2)")                        alter_parts.append(f"ADD COLUMN `{code}_high` DECIMAL(10,2)")                        alter_parts.append(f"ADD COLUMN `{code}_low` DECIMAL(10,2)")                        alter_parts.append(f"ADD COLUMN `{code}_close` DECIMAL(10,2)")                        alter_parts.append(f"ADD COLUMN `{code}_volume` BIGINT")                        alter_parts.append(f"ADD COLUMN `{code}_amount` DECIMAL(20,2)")                    alter_sql = f"ALTER TABLE `{last_table}` {', '.join(alter_parts)}"                    cursor.execute(alter_sql)                    conn.commit()# 更新映射                    table_mapping[last_table].extend(stocks_to_add)                    current_new_stock_idx += len(stocks_to_add)# 2. 如果还有剩余新股票，创建新表while current_new_stock_idx < len(new_stocks):            next_batch_idx = len(tables) + 1            new_table_name = f"{table_prefix}{next_batch_idx}"            stocks_for_new_table = new_stocks[current_new_stock_idx : current_new_stock_idx + STOCKS_PER_TABLE]            print(f"  创建新表 {new_table_name}，包含 {len(stocks_for_new_table)} 只代码...")# 构建列定义            columns_def = ["id INT AUTO_INCREMENT PRIMARY KEY", "date DATE NOT NULL"]for code in stocks_for_new_table:                columns_def.append(f"`{code}_open` DECIMAL(10,2)")                columns_def.append(f"`{code}_high` DECIMAL(10,2)")                columns_def.append(f"`{code}_low` DECIMAL(10,2)")                columns_def.append(f"`{code}_close` DECIMAL(10,2)")                columns_def.append(f"`{code}_volume` BIGINT")                columns_def.append(f"`{code}_amount` DECIMAL(20,2)")            create_sql = f"""            CREATE TABLE `{new_table_name}` ({', '.join(columns_def)},                UNIQUE KEY uk_date (date)            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='板块日线数据'            """            cursor.execute(create_sql)            conn.commit()# 更新状态            tables.append(new_table_name)            table_mapping[new_table_name] = stocks_for_new_table            current_new_stock_idx += len(stocks_for_new_table)return table_mapping, new_stocksexcept Exception as e:        print(f"更新数据库结构失败: {e}")        conn.rollback()raisefinally:if cursor:            cursor.close()# ==================== 收集数据 ====================defcollect_data(reader, stock_config):"""    收集数据    stock_config: dict, {stock_code: start_date}    如果 start_date 为 None，则获取全量数据    如果 start_date 有值，则获取该日期之后的数据    """    all_data = {}    failed_stocks = []for code in tqdm(stock_config.keys(), desc="  读取本地行情", leave=False):try:            day_file = resolve_day_file_path(code)            df = parse_tdx_day_file(day_file) if day_file elseNoneif df isNoneor df.empty:                df = reader.daily(symbol=code)if df isnotNoneandnot df.empty:                    df = df.reset_index()if df isNoneor df.empty:                failed_stocks.append(code)continueifnot isinstance(df.index, pd.DatetimeIndex) and'date'notin df.columns:                failed_stocks.append(code)continueif'date'notin df.columns:                df = df.reset_index()            start_date = stock_config[code]if start_date:                ts_start_date = pd.Timestamp(start_date)                df = df[df['date'] >= ts_start_date]if df.empty:continuefor _, row in df.iterrows():                date = row['date'].strftime('%Y-%m-%d')if date notin all_data:                    all_data[date] = {'date': date}                all_data[date][f'{code}_open'] = float(row['open']) if pd.notna(row['open']) elseNone                all_data[date][f'{code}_high'] = float(row['high']) if pd.notna(row['high']) elseNone                all_data[date][f'{code}_low'] = float(row['low']) if pd.notna(row['low']) elseNone                all_data[date][f'{code}_close'] = float(row['close']) if pd.notna(row['close']) elseNone                all_data[date][f'{code}_volume'] = int(row['volume']) if pd.notna(row['volume']) elseNone                all_data[date][f'{code}_amount'] = float(row['amount']) if pd.notna(row['amount']) elseNoneexcept Exception:            failed_stocks.append(code)continuereturn all_data, failed_stocks# ==================== 插入/更新数据 ====================defupsert_data(conn, table_name, stock_codes, data_dict):"""插入或更新数据"""    cursor = Nonetry:        cursor = conn.cursor()        inserted = 0        updated = 0for date in sorted(data_dict.keys()):            row_data = data_dict[date]# 检查该日期是否已存在            cursor.execute(f"SELECT COUNT(*) FROM `{table_name}` WHERE date = %s", (date,))            exists = cursor.fetchone()[0] > 0if exists:# 更新已有记录                set_parts = []                values = []                has_update = Falsefor code in stock_codes:# 只有当该股票在该日期有数据时才更新iff'{code}_open'in row_data:for field in ['open', 'high', 'low', 'close', 'volume', 'amount']:                            col_name = f'{code}_{field}'                            set_parts.append(f"`{col_name}` = %s")                            values.append(row_data.get(col_name))                        has_update = Trueif has_update:                    values.append(date)  # WHERE date = %s                    update_sql = f"UPDATE `{table_name}` SET {', '.join(set_parts)} WHERE date = %s"                    cursor.execute(update_sql, values)                    updated += 1else:# 插入新记录                values = [date]# 构建列名                columns = ['date']for code in stock_codes:                    columns.extend([f'{code}_open', f'{code}_high', f'{code}_low', f'{code}_close', f'{code}_volume', f'{code}_amount'                    ])                    values.append(row_data.get(f'{code}_open'))                    values.append(row_data.get(f'{code}_high'))                    values.append(row_data.get(f'{code}_low'))                    values.append(row_data.get(f'{code}_close'))                    values.append(row_data.get(f'{code}_volume'))                    values.append(row_data.get(f'{code}_amount'))                placeholders = ', '.join(['%s'] * len(columns))                column_names = ', '.join([f'`{col}`'for col in columns])                insert_sql = f"INSERT INTO `{table_name}` ({column_names}) VALUES ({placeholders})"                cursor.execute(insert_sql, values)                inserted += 1        conn.commit()return inserted, updatedexcept Exception as e:        print(f"插入/更新数据失败: {e}")        conn.rollback()raisefinally:if cursor:            cursor.close()# ==================== 获取数据库最新日期 ====================defget_latest_date_in_db(conn, table_prefix):"""获取数据库中最新的交易日期"""    cursor = Nonetry:        cursor = conn.cursor()        cursor.execute(f"SHOW TABLES LIKE '{table_prefix}%'")        tables = [row[0] for row in cursor.fetchall()]ifnot tables:returnNone# 从第一个表获取最新日期        cursor.execute(f"SELECT MAX(date) FROM `{tables[0]}`")        result = cursor.fetchone()if result and result[0]:return result[0]returnNonefinally:if cursor:            cursor.close()# ==================== 处理同步任务 ====================defprocess_sync_task(conn, reader, block_file, table_prefix, task_name):    print("\n" + "="*60)    print(f"开始任务: {task_name}")    print(f"板块文件: {block_file}")    print(f"数据表前缀: {table_prefix}")    print("="*60)# 解析板块文件    print(f"\n正在读取板块文件: {block_file}")ifnot os.path.exists(block_file):        print(f"文件不存在: {block_file}，跳过该任务")return    all_stocks_from_file = parse_block_file(block_file)    print(f"板块文件中共有 {len(all_stocks_from_file)} 只代码")ifnot all_stocks_from_file:        print("未找到代码，跳过")return# 更新数据库结构    table_mapping, new_stocks = update_schema(conn, all_stocks_from_file, table_prefix)# 获取全局最新日期    latest_db_date = get_latest_date_in_db(conn, table_prefix)    print(f"\n数据库现有最新日期: {latest_db_date}")# 逐表处理    total_inserted = 0    total_updated = 0    print("\n开始数据更新...")for table_name, stock_codes in table_mapping.items():        print(f"\n处理表: {table_name} ({len(stock_codes)} 只代码)")# 构建配置：新股全量，旧股增量        stock_config = {}        new_cnt = 0        old_cnt = 0for code in stock_codes:if code in new_stocks:                stock_config[code] = None# 全量                new_cnt += 1else:                stock_config[code] = latest_db_date # 增量                old_cnt += 1        print(f"  包含: {new_cnt} 只新代码 (全量), {old_cnt} 只旧代码 (增量)")# 收集数据        print(f"  正在收集数据...")        batch_data, failed = collect_data(reader, stock_config)ifnot batch_data:            print(f"  无数据需要更新")continue        dates = sorted(batch_data.keys())        print(f"  收集到 {len(dates)} 个交易日数据: {dates[0]} ~ {dates[-1]}")# 插入/更新        print(f"  正在写入数据库...")        inserted, updated = upsert_data(conn, table_name, stock_codes, batch_data)        total_inserted += inserted        total_updated += updated        print(f"  ✓ 插入 {inserted} 行，更新 {updated} 行")if failed:            print(f"  ! 失败代码: {len(failed)} 只")    print(f"\n{task_name} 任务完成！总插入: {total_inserted} 行, 总更新: {total_updated} 行")# ==================== 主函数 ====================defmain():    print("="*60)    print("通达信数据智能同步工具 (支持多板块)")    print("="*60)# 1. 连接数据库    conn = connect_to_database()    print("数据库连接成功")try:# 2. 初始化 Reader        print("\n正在初始化 Reader...")        reader = Reader.factory(market='std', tdxdir=TDX_DIR)# 3. 执行任务# 任务1: 股票 (QABJ) -> stock_batch_        process_sync_task(conn, reader, BLOCK_FILE, 'stock_batch_', '股票行情(QABJ)')# 任务2: ETF (ETFSJ) -> etf_batch_        etf_block_file = r'D:\交易\通达信\T0002\blocknew\ETFSJ.blk'        process_sync_task(conn, reader, etf_block_file, 'etf_batch_', 'ETF行情(ETFSJ)')except Exception as e:        print(f"\n程序运行出错: {e}")import traceback        traceback.print_exc()finally:        conn.close()        print("\n数据库连接已关闭")if __name__ == "__main__":    main()