--- name: a-stock-get description: Specialized A-share stock data collector. Automatically fetch and store daily/weekly/monthly historical K-line data for all A-share stocks in SQLite database. --- # A-Share-Get / A股数据获取工具 **English** | **中文** --- ## Overview / 概述 **English:** A specialized data collection tool for Chinese A-share market. Automatically fetches and stores stock list, daily, weekly, and monthly historical K-line data into a local SQLite database. Designed specifically for quantitative analysts who need a complete local copy of A-share market data. **中文:** 专门针对中国A股市场的数据获取工具。自动采集并存储股票列表、日线、周线、月线历史K线数据到本地SQLite数据库。专为需要完整A股本地数据副本的量化分析师设计。 --- ## Features / 功能特性 **English:** - **Automated Stock List Management**: Fetches and updates all tradable stocks from A-share markets (60* Shanghai, 30* ChiNext, 00* Shenzhen) - **Multi-Frequency Data Collection**: Supports parallel fetching for daily, weekly, and monthly K-line data - **SQLite Database Storage**: Local persistent storage, easy to query for quantitative analysis - **Automatic Filtering**: Excludes delisted and pre-IPO stocks automatically - **Data Integrity**: Tracks last fetch timestamp for each frequency, supports incremental updates - **Multiple Data Sources**: EastMoney/Sina/Tencent with automatic failover **中文:** - **自动化股票列表管理**: 获取并更新A股市场全部可交易股票（沪市60*、创业板30*、深市00*） - **多频率数据采集**: 支持并行获取日线、周线、月线K线数据 - **SQLite数据库存储**: 本地持久化存储，便于量化分析查询 - **自动过滤**: 自动排除退市和未上市股票 - **数据完整性**: 记录每种频率最后获取时间戳，支持增量更新 - **多数据源冗余**: 东方财富/新浪/腾讯，自动故障切换 --- ## Database Schema / 数据库架构 ### Stock List Table (stocks) | Column / 字段 | Type / 类型 | Description / 说明 | |---------------|-------------|-------------------| | code | TEXT | Stock code (e.g., 600519) / 股票代码 | | name | TEXT | Stock name (e.g., 贵州茅台) / 股票名称 | | market | TEXT | Market type (60/30/00) / 市场类型 | | day_get | TIMESTAMP | Last daily data fetch time / 最后日线数据获取时间 | | week_get | TIMESTAMP | Last weekly data fetch time / 最后周线数据获取时间 | | month_get | TIMESTAMP | Last monthly data fetch time / 最后月线数据获取时间 | | status | TEXT | Stock status (active/delisted) / 股票状态 | | created_at | TIMESTAMP | Record creation time / 记录创建时间 | --- ## Installation & Setup / 安装与设置 ### Step 1: Database Initialization / 第一步：数据库初始化 **English:** Run the initialization script to create the database and tables: ```bash python scripts/init_db.py ``` **中文:** 运行初始化脚本创建数据库和表： ```bash python scripts/init_db.py ``` ### Step 2: Fetch Stock List / 第二步：获取股票列表 **English:** Fetch all tradable stocks from A-share markets: ```bash python scripts/fetch_stocks.py ``` **中文:** 从A股市场获取所有可交易股票列表： ```bash python scripts/fetch_stocks.py ``` ### Step 3: Start Data Collection / 第三步：启动数据收集 **English:** ```bash # Enhanced data fetching with external events python scripts/day.py get all --limit 10 python scripts/week.py get all --limit 10 python scripts/month.py get all --limit 10 # Traditional usage (fetch all active stocks) python scripts/day.py python scripts/week.py python scripts/month.py # Database reset and fetch tool python scripts/db_reset.py reset status python scripts/db_reset.py fetch day 000001 # Parallel fetching for large batches python scripts/day_parallel.py python scripts/week_parallel.py python scripts/month_parallel.py ``` **中文:** ```bash # 增强版数据获取（支持外部事件） python scripts/day.py get all --limit 10 python scripts/week.py get all --limit 10 python scripts/month.py get all --limit 10 # 传统用法（获取所有活跃股票） python scripts/day.py python scripts/week.py python scripts/month.py # 数据库重置与获取工具 python scripts/db_reset.py reset status python scripts/db_reset.py fetch day 000001 # 并行获取大量数据 python scripts/day_parallel.py python scripts/week_parallel.py python scripts/month_parallel.py ``` --- ## File Structure / 文件结构 ``` a-stock-get/ ├── SKILL.md # This documentation ├── scripts/ │ ├── init_db.py # Database initialization │ ├── fetch_stocks.py # Fetch stock list from API │ ├── day.py # Enhanced daily data fetch with external events │ ├── day_original.py # Original daily data fetch (backup) │ ├── day_parallel.py # Parallel daily data fetch │ ├── db_reset.py # Database reset and fetch tool │ ├── week.py # Enhanced weekly data fetch with external events │ ├── week_original.py # Original weekly data fetch (backup) │ ├── week_parallel.py # Parallel weekly data fetch │ ├── month.py # Enhanced monthly data fetch with external events │ ├── month_original.py # Original monthly data fetch (backup) │ ├── month_parallel.py # Parallel monthly data fetch │ ├── data_validation.py # Data validation and integrity check │ ├── data_repair.py # Data repair tool │ ├── schedule_config.py # OpenClaw cron job configuration │ └── README.md # Detailed usage documentation ├── references/ │ └── data_sources.md # Data source documentation └── D:\xistock\ # Data directory (external) └── stock.db # SQLite database ``` --- ## Data Sources / 数据来源 **English:** - **East Money API**: Real-time stock quotes and listing information - **Sina Finance**: Historical data for technical analysis - **Tencent Finance**: Alternative data source for redundancy **中文:** - **东方财富API**: 实时股票行情和上市信息 - **新浪财经**: 技术分析历史数据 - **腾讯财经**: 冗余备用数据源 --- ## Requirements / 依赖要求 ```bash pip install requests pip install sqlite3 pip install pandas pip install akshare # Chinese stock data library ``` --- ## Notes / 注意事项 **English:** - Database file is stored at `D:\xistock\stock.db` for data persistence - Stock list should be updated regularly (e.g., weekly) to capture new listings and delistings - API rate limits apply when fetching data; parallel mode improves speed - This system is for research and educational purposes; comply with local regulations for actual trading **中文:** - 数据库文件存储在 `D:\xistock\stock.db` 确保数据持久化 - 股票列表应定期更新（如每周）以捕捉新股上市和退市变化 - 获取数据受API速率限制，并行模式提升速度 - 本系统仅用于研究和教育目的；实际交易请遵守当地法规 ## Todo / 待办事项 - [x] Add incremental update mode (only fetch stocks not updated today) - [x] Add external event control to day.py, week.py, and month.py - [x] Create database reset and fetch tool (db_reset.py) - [x] Add data validation and integrity checks (data_validation.py) - [x] Add data repair tool (data_repair.py) - [x] Add OpenClaw cron job configuration (schedule_config.py) - [ ] Integrate with OpenClaw heartbeat for scheduled automatic updates - [ ] Add advanced analytics and reporting features --- ## References / 参考资料 - [AkShare Documentation](https://akshare.readthedocs.io/) - Chinese financial data library - [SQLite Python Tutorial](https://docs.python.org/3/library/sqlite3.html) --- **Version**: 1.0.0 **Last Updated**: 2026-03-13 **Author**: jakey **Email**: zhuxi0906@gmail.com **Wechat**: jakeycis FILE:scripts/data_repair.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Data Repair Tool for XI Stock System 羲股票监控系统数据修复工具 This script repairs common issues found in the stock database and data files. 本脚本修复股票数据库和数据文件中发现的常见问题。 """ import os import sys import sqlite3 import shutil from datetime import datetime, timedelta import argparse # Fix encoding for Windows console if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" DATA_DIRS = { "day": "D:\\xistock\\day", "week": "D:\\xistock\\week", "month": "D:\\xistock\\month" } BACKUP_DIR = "D:\\xistock\\backup" def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- 数据库未找到: {DB_PATH}") print(" 请先运行 init_db.py 和 fetch_stocks.py!") exit(1) return sqlite3.connect(DB_PATH) def create_backup(): """Create backup of database before repair""" print("=" * 70) print("创建数据库备份") print("=" * 70) if not os.path.exists(BACKUP_DIR): os.makedirs(BACKUP_DIR) timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') backup_file = os.path.join(BACKUP_DIR, f"stock_backup_{timestamp}.db") try: shutil.copy2(DB_PATH, backup_file) print(f"✅ 数据库备份已创建: {backup_file}") print(f" 原始文件: {DB_PATH}") print(f" 备份大小: {os.path.getsize(backup_file):,} 字节") return backup_file except Exception as e: print(f"❌ 备份创建失败: {e}") return None def repair_database_structure(): """Repair database structure issues""" print("\n" + "=" * 70) print("修复数据库结构") print("=" * 70) conn = get_db_connection() cursor = conn.cursor() repairs = [] try: # 1. Check if table exists, create if not cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='stocks'") if not cursor.fetchone(): print("⚠️ 表 'stocks' 不存在，正在创建...") cursor.execute(''' CREATE TABLE stocks ( code TEXT PRIMARY KEY, name TEXT NOT NULL, market TEXT NOT NULL, day_get TIMESTAMP, week_get TIMESTAMP, month_get TIMESTAMP, status TEXT DEFAULT 'active', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') repairs.append("✅ 创建表 'stocks'") # 2. Check and add missing columns cursor.execute("PRAGMA table_info(stocks)") existing_columns = [row[1] for row in cursor.fetchall()] required_columns = [ ('code', 'TEXT PRIMARY KEY'), ('name', 'TEXT NOT NULL'), ('market', 'TEXT NOT NULL'), ('day_get', 'TIMESTAMP'), ('week_get', 'TIMESTAMP'), ('month_get', 'TIMESTAMP'), ('status', 'TEXT DEFAULT "active"'), ('created_at', 'TIMESTAMP DEFAULT CURRENT_TIMESTAMP') ] for col_name, col_type in required_columns: if col_name not in existing_columns: print(f"⚠️ 缺少列 '{col_name}'，正在添加...") try: cursor.execute(f"ALTER TABLE stocks ADD COLUMN {col_name} {col_type}") repairs.append(f"✅ 添加列 '{col_name}'") except Exception as e: repairs.append(f"❌ 添加列 '{col_name}' 失败: {e}") # 3. Create indexes if missing indexes = [ ("idx_market", "CREATE INDEX idx_market ON stocks(market)"), ("idx_status", "CREATE INDEX idx_status ON stocks(status)"), ("idx_day_get", "CREATE INDEX idx_day_get ON stocks(day_get)"), ("idx_week_get", "CREATE INDEX idx_week_get ON stocks(week_get)"), ("idx_month_get", "CREATE INDEX idx_month_get ON stocks(month_get)") ] for idx_name, idx_sql in indexes: cursor.execute(f"SELECT name FROM sqlite_master WHERE type='index' AND name='{idx_name}'") if not cursor.fetchone(): print(f"⚠️ 缺少索引 '{idx_name}'，正在创建...") try: cursor.execute(idx_sql) repairs.append(f"✅ 创建索引 '{idx_name}'") except Exception as e: repairs.append(f"❌ 创建索引 '{idx_name}' 失败: {e}") conn.commit() except Exception as e: repairs.append(f"❌ 数据库结构修复错误: {e}") conn.rollback() finally: conn.close() return repairs def repair_duplicate_stocks(): """Remove duplicate stock entries""" print("\n" + "=" * 70) print("修复重复股票记录") print("=" * 70) conn = get_db_connection() cursor = conn.cursor() repairs = [] try: # Find duplicates cursor.execute(""" SELECT code, COUNT(*) as cnt, GROUP_CONCAT(rowid) as ids FROM stocks GROUP BY code HAVING COUNT(*) > 1 """) duplicates = cursor.fetchall() if not duplicates: repairs.append("✅ 无重复股票记录") return repairs print(f"发现 {len(duplicates)} 个重复股票代码") for code, count, ids in duplicates: id_list = ids.split(',') keep_id = id_list[0] # Keep the first record delete_ids = id_list[1:] # Delete the rest print(f"🔄 处理重复股票: {code} (出现 {count} 次)") print(f" 保留记录: {keep_id}") print(f" 删除记录: {', '.join(delete_ids)}") # Delete duplicate records placeholders = ','.join(['?'] * len(delete_ids)) cursor.execute(f"DELETE FROM stocks WHERE rowid IN ({placeholders})", delete_ids) repairs.append(f"✅ 删除重复股票 {code}: 保留1条，删除{len(delete_ids)}条") conn.commit() except Exception as e: repairs.append(f"❌ 重复记录修复错误: {e}") conn.rollback() finally: conn.close() return repairs def repair_missing_timestamps(): """Set NULL timestamps for old records""" print("\n" + "=" * 70) print("修复缺失的时间戳") print("=" * 70) conn = get_db_connection() cursor = conn.cursor() repairs = [] current_time = datetime.now() try: # Define thresholds for each frequency thresholds = [ ('day_get', timedelta(days=1)), ('week_get', timedelta(days=7)), ('month_get', timedelta(days=30)) ] for field, threshold in thresholds: # Count records with NULL or old timestamps cursor.execute(f""" SELECT COUNT(*) FROM stocks WHERE {field} IS NULL OR {field} < ? """, (current_time - threshold,)) count = cursor.fetchone()[0] if count > 0: print(f"🔄 修复 {field}: {count} 条记录需要更新") # Set NULL for old timestamps cursor.execute(f""" UPDATE stocks SET {field} = NULL WHERE {field} < ? """, (current_time - threshold,)) repairs.append(f"✅ 修复 {field}: 重置 {count} 条过时记录") conn.commit() except Exception as e: repairs.append(f"❌ 时间戳修复错误: {e}") conn.rollback() finally: conn.close() return repairs def repair_data_files(): """Repair data file issues""" print("\n" + "=" * 70) print("修复数据文件") print("=" * 70) repairs = [] for frequency, data_dir in DATA_DIRS.items(): print(f"\n检查 {frequency}线数据目录: {data_dir}") if not os.path.exists(data_dir): print(f"⚠️ 目录不存在，正在创建: {data_dir}") os.makedirs(data_dir) repairs.append(f"✅ 创建目录: {data_dir}") continue # Check for empty files empty_files = [] corrupted_files = [] for filename in os.listdir(data_dir): if not filename.endswith('.txt'): continue filepath = os.path.join(data_dir, filename) try: # Check if file is empty if os.path.getsize(filepath) == 0: empty_files.append(filename) continue # Check file content with open(filepath, 'r', encoding='utf-8') as f: lines = f.readlines() if len(lines) < 2: # Less than header + 1 data line corrupted_files.append(filename) else: # Check header if lines[0].strip() != "date,open,close,high,low,change_pct": corrupted_files.append(filename) except Exception: corrupted_files.append(filename) # Report and repair if empty_files: print(f"⚠️ 发现 {len(empty_files)} 个空文件") for filename in empty_files[:5]: # Show first 5 print(f" - {filename}") # Option to delete empty files if len(empty_files) > 0: repair = input(f"\n删除 {len(empty_files)} 个空文件? (y/n): ") if repair.lower() == 'y': for filename in empty_files: filepath = os.path.join(data_dir, filename) os.remove(filepath) repairs.append(f"✅ 删除 {len(empty_files)} 个空文件 ({frequency}线)") if corrupted_files: print(f"⚠️ 发现 {len(corrupted_files)} 个损坏文件") for filename in corrupted_files[:5]: print(f" - {filename}") # Option to delete corrupted files if len(corrupted_files) > 0: repair = input(f"\n删除 {len(corrupted_files)} 个损坏文件? (y/n): ") if repair.lower() == 'y': for filename in corrupted_files: filepath = os.path.join(data_dir, filename) os.remove(filepath) repairs.append(f"✅ 删除 {len(corrupted_files)} 个损坏文件 ({frequency}线)") if not empty_files and not corrupted_files: repairs.append(f"✅ {frequency}线数据文件正常") return repairs def sync_database_with_files(): """Sync database with existing data files""" print("\n" + "=" * 70) print("同步数据库与数据文件") print("=" * 70) repairs = [] conn = get_db_connection() cursor = conn.cursor() try: # Get all stocks from database cursor.execute("SELECT code, name FROM stocks WHERE status = 'active'") db_stocks = {row[0]: row[1] for row in cursor.fetchall()} # Check each frequency directory for frequency, data_dir in DATA_DIRS.items(): if not os.path.exists(data_dir): continue files = [f for f in os.listdir(data_dir) if f.endswith('.txt')] file_codes = set() for filename in files: # Extract code from filename try: if '_' in filename: code = filename.split('_')[-1].replace('.txt', '') file_codes.add(code) except: pass # Find stocks in database but missing files missing_files = set(db_stocks.keys()) - file_codes if missing_files: print(f"🔄 {frequency}线: {len(missing_files)} 只股票缺少数据文件") # Mark these stocks as needing update placeholders = ','.join(['?'] * len(missing_files)) cursor.execute(f""" UPDATE stocks SET {frequency}_get = NULL WHERE code IN ({placeholders}) """, list(missing_files)) repairs.append(f"✅ 标记 {len(missing_files)} 只股票需要更新{frequency}线数据") conn.commit() except Exception as e: repairs.append(f"❌ 同步修复错误: {e}") conn.rollback() finally: conn.close() return repairs def generate_repair_report(repairs, backup_file=None): """Generate repair report""" print("\n" + "=" * 70) print("修复报告") print("=" * 70) if backup_file: print(f"📂 备份文件: {backup_file}") if not repairs: print("✅ 无需修复，系统状态良好") return print(f"完成 {len(repairs)} 项修复:") print("-" * 70) for i, repair in enumerate(repairs, 1): print(f"{i}. {repair}") print("-" * 70) successful = len([r for r in repairs if r.startswith('✅')]) failed = len([r for r in repairs if r.startswith('❌')]) print(f"✅ 成功: {successful} 项") if failed > 0: print(f"❌ 失败: {failed} 项") # Save report report_file = f"repair_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" with open(report_file, 'w', encoding='utf-8') as f: f.write(f"XI Stock System Repair Report\n") f.write(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"备份文件: {backup_file or '无'}\n") f.write(f"修复项数: {len(repairs)}\n\n") if repairs: f.write("修复记录:\n") for i, repair in enumerate(repairs, 1): f.write(f"{i}. {repair}\n") f.write(f"\n总结:\n") f.write(f" 成功: {successful} 项\n") f.write(f" 失败: {failed} 项\n") print(f"\n📄 修复报告已保存: {report_file}") def main(): """Main function""" parser = argparse.ArgumentParser( description='Data Repair Tool for XI Stock System', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 使用示例: data_repair.py --fix-all - 修复所有问题（推荐） data_repair.py --fix-db - 只修复数据库问题 data_repair.py --fix-files - 只修复数据文件问题 data_repair.py --fix-timestamps - 只修复时间戳问题 data_repair.py --fix-duplicates - 只修复重复记录安全提示: 1. 修复前会自动创建数据库备份 2. 数据文件修复需要确认 3. 建议先运行 data_validation.py 检查问题 """ ) parser.add_argument('--fix-all', action='store_true', help='修复所有问题') parser.add_argument('--fix-db', action='store_true', help='修复数据库结构问题') parser.add_argument('--fix-files', action='store_true', help='修复数据文件问题') parser.add_argument('--fix-timestamps', action='store_true', help='修复时间戳问题') parser.add_argument('--fix-duplicates', action='store_true', help='修复重复记录问题') parser.add_argument('--fix-sync', action='store_true', help='修复数据库与文件同步问题') parser.add_argument('--no-backup', action='store_true', help='不创建备份（不推荐）') args = parser.parse_args() # If no specific fix specified, default to all if not any([args.fix_all, args.fix_db, args.fix_files, args.fix_timestamps, args.fix_duplicates, args.fix_sync]): args.fix_all = True all_repairs = [] try: print("=" * 70) print("XI Stock System Data Repair Tool") print("羲股票监控系统数据修复工具") print("=" * 70) # Create backup unless disabled backup_file = None if not args.no_backup: backup_file = create_backup() if not backup_file: print("⚠️ 备份失败，继续修复? (y/n): ") if input().lower() != 'y': return 1 else: print("⚠️ 警告: 未创建备份，直接进行修复") confirm = input("继续? (y/n): ") if confirm.lower() != 'y': return 1 # Perform repairs based on arguments if args.fix_all or args.fix_db: repairs = repair_database_structure() all_repairs.extend(repairs) if args.fix_all or args.fix_duplicates: repairs = repair_duplicate_stocks() all_repairs.extend(repairs) if args.fix_all or args.fix_timestamps: repairs = repair_missing_timestamps() all_repairs.extend(repairs) if args.fix_all or args.fix_files: repairs = repair_data_files() all_repairs.extend(repairs) if args.fix_all or args.fix_sync: repairs = sync_database_with_files() all_repairs.extend(repairs) # Generate report generate_repair_report(all_repairs, backup_file) print("\n🎉 修复完成!") print("建议运行以下命令验证修复效果:") print(" python data_validation.py all") except KeyboardInterrupt: print("\n\n- 用户中断") return 1 except Exception as e: print(f"\n- 错误: {e}") import traceback traceback.print_exc() return 1 return 0 if __name__ == "__main__": exit(main()) FILE:scripts/data_validation.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Data Validation and Integrity Check for XI Stock System 羲股票监控系统数据验证与完整性检查 This script validates data integrity and checks for issues in the stock database and data files. 本脚本验证数据完整性并检查股票数据库和数据文件的问题。 """ import os import sys import sqlite3 import json import csv from datetime import datetime, timedelta import argparse # Fix encoding for Windows console if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" DATA_DIRS = { "day": "D:\\xistock\\day", "week": "D:\\xistock\\week", "month": "D:\\xistock\\month" } def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- 数据库未找到: {DB_PATH}") print(" 请先运行 init_db.py 和 fetch_stocks.py!") exit(1) return sqlite3.connect(DB_PATH) def check_database_integrity(): """Check database integrity and structure""" print("=" * 70) print("数据库完整性检查") print("=" * 70) conn = get_db_connection() cursor = conn.cursor() issues = [] try: # 1. Check if table exists cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='stocks'") if not cursor.fetchone(): issues.append("❌ 表 'stocks' 不存在") else: print("✅ 表 'stocks' 存在") # 2. Check table structure cursor.execute("PRAGMA table_info(stocks)") columns = [row[1] for row in cursor.fetchall()] required_columns = ['code', 'name', 'market', 'day_get', 'week_get', 'month_get', 'status', 'created_at'] for col in required_columns: if col not in columns: issues.append(f"❌ 缺少列: {col}") else: print(f"✅ 列 '{col}' 存在") # 3. Check total records cursor.execute("SELECT COUNT(*) FROM stocks") total = cursor.fetchone()[0] print(f"📊 总股票记录: {total}") if total == 0: issues.append("⚠️ 数据库中没有股票记录") # 4. Check active stocks cursor.execute("SELECT COUNT(*) FROM stocks WHERE status = 'active'") active = cursor.fetchone()[0] print(f"📈 活跃股票: {active} ({active/total*100:.1f}%)") # 5. Check timestamp fields timestamp_fields = ['day_get', 'week_get', 'month_get'] for field in timestamp_fields: cursor.execute(f"SELECT COUNT(*) FROM stocks WHERE {field} IS NOT NULL") updated = cursor.fetchone()[0] print(f"🕒 {field}: {updated} 已更新 ({updated/total*100:.1f}%)") # 6. Check for duplicate codes cursor.execute("SELECT code, COUNT(*) FROM stocks GROUP BY code HAVING COUNT(*) > 1") duplicates = cursor.fetchall() if duplicates: for code, count in duplicates: issues.append(f"❌ 重复股票代码: {code} (出现 {count} 次)") else: print("✅ 无重复股票代码") # 7. Check for invalid market codes cursor.execute("SELECT DISTINCT market FROM stocks") markets = [row[0] for row in cursor.fetchall()] print(f"🏢 市场类型: {', '.join(markets)}") invalid_markets = [m for m in markets if m not in ['00', '30', '60']] if invalid_markets: issues.append(f"❌ 无效市场类型: {invalid_markets}") except Exception as e: issues.append(f"❌ 数据库检查错误: {e}") finally: conn.close() return issues def check_data_files(frequency): """Check data files for specific frequency""" print(f"\n{'='*70}") print(f"{frequency}线数据文件检查") print(f"{'='*70}") data_dir = DATA_DIRS.get(frequency) if not data_dir or not os.path.exists(data_dir): return [f"❌ 数据目录不存在: {data_dir or frequency}"] issues = [] try: # Count files files = [f for f in os.listdir(data_dir) if f.endswith('.txt')] print(f"📁 数据文件数量: {len(files)}") if len(files) == 0: issues.append(f"⚠️ {frequency}线数据目录为空") return issues # Check first 5 files for format sample_files = files[:5] if len(files) > 5 else files for filename in sample_files: filepath = os.path.join(data_dir, filename) try: with open(filepath, 'r', encoding='utf-8') as f: # Check header header = f.readline().strip() if header != "date,open,close,high,low,change_pct": issues.append(f"❌ 文件头格式错误: {filename}") continue # Check data lines lines = f.readlines() if not lines: issues.append(f"⚠️ 空数据文件: {filename}") continue # Check line format for i, line in enumerate(lines[:10], 1): # Check first 10 lines parts = line.strip().split(',') if len(parts) != 6: issues.append(f"❌ 数据行格式错误 {filename}:{i}") break # Check if values can be parsed try: date_str = parts[0] float(parts[1]) # open float(parts[2]) # close float(parts[3]) # high float(parts[4]) # low float(parts[5]) # change_pct except ValueError: issues.append(f"❌ 数据值解析错误 {filename}:{i}") break print(f"✅ 文件检查通过: {filename}") except Exception as e: issues.append(f"❌ 文件读取错误 {filename}: {e}") # Check file naming consistency for filename in files[:20]: # Check first 20 files if '_' not in filename: issues.append(f"❌ 文件名格式错误: {filename}") break print(f"✅ 文件命名格式检查通过") except Exception as e: issues.append(f"❌ 数据文件检查错误: {e}") return issues def check_database_files_sync(): """Check synchronization between database and data files""" print(f"\n{'='*70}") print("数据库与文件同步检查") print(f"{'='*70}") issues = [] conn = get_db_connection() cursor = conn.cursor() try: # Get all active stocks from database cursor.execute("SELECT code, name FROM stocks WHERE status = 'active'") db_stocks = {row[0]: row[1] for row in cursor.fetchall()} # Check each frequency directory for frequency, data_dir in DATA_DIRS.items(): if not os.path.exists(data_dir): print(f"⚠️ 目录不存在: {data_dir}") continue files = [f for f in os.listdir(data_dir) if f.endswith('.txt')] file_codes = set() for filename in files: # Extract code from filename (format: 名称_代码.txt) try: if '_' in filename: code = filename.split('_')[-1].replace('.txt', '') file_codes.add(code) except: pass # Find missing files missing_files = set(db_stocks.keys()) - file_codes if missing_files: sample_missing = list(missing_files)[:5] issues.append(f"⚠️ {frequency}线缺失文件: {len(missing_files)} 只股票 (示例: {', '.join(sample_missing)})") # Find extra files (not in database) extra_files = file_codes - set(db_stocks.keys()) if extra_files: sample_extra = list(extra_files)[:5] issues.append(f"⚠️ {frequency}线多余文件: {len(extra_files)} 个文件 (示例: {', '.join(sample_extra)})") print(f"📊 {frequency}线: 数据库 {len(db_stocks)} 只股票, 文件 {len(files)} 个") except Exception as e: issues.append(f"❌ 同步检查错误: {e}") finally: conn.close() return issues def check_timestamp_consistency(): """Check timestamp consistency and freshness""" print(f"\n{'='*70}") print("时间戳一致性检查") print(f"{'='*70}") issues = [] conn = get_db_connection() cursor = conn.cursor() try: current_time = datetime.now() # Check each timestamp field timestamp_fields = [ ('day_get', '日线', timedelta(days=1)), ('week_get', '周线', timedelta(days=7)), ('month_get', '月线', timedelta(days=30)) ] for field, name, threshold in timestamp_fields: # Get oldest timestamp cursor.execute(f""" SELECT code, name, {field} FROM stocks WHERE {field} IS NOT NULL ORDER BY {field} ASC LIMIT 5 """) oldest = cursor.fetchall() if oldest: oldest_date = datetime.strptime(oldest[0][2], '%Y-%m-%d %H:%M:%S') age_days = (current_time - oldest_date).days print(f"📅 {name}最旧数据: {oldest[0][0]} {oldest[0][1]} ({oldest_date})") print(f" 数据年龄: {age_days} 天") if age_days > threshold.days: issues.append(f"⚠️ {name}数据过旧: {age_days} 天") # Count stocks needing update cursor.execute(f"SELECT COUNT(*) FROM stocks WHERE {field} IS NULL OR {field} < ?", (current_time - threshold,)) need_update = cursor.fetchone()[0] print(f"🔄 需要更新{name}: {need_update} 只股票") except Exception as e: issues.append(f"❌ 时间戳检查错误: {e}") finally: conn.close() return issues def generate_report(issues): """Generate validation report""" print(f"\n{'='*70}") print("验证报告") print(f"{'='*70}") if not issues: print("🎉 所有检查通过！系统健康状态良好。") return True print(f"发现 {len(issues)} 个问题:") print("-" * 70) for i, issue in enumerate(issues, 1): print(f"{i}. {issue}") print("-" * 70) # Categorize issues critical = [i for i in issues if i.startswith('❌')] warning = [i for i in issues if i.startswith('⚠️')] if critical: print(f"❌ 严重问题: {len(critical)} 个") print(" 建议立即处理这些问题") if warning: print(f"⚠️ 警告: {len(warning)} 个") print(" 建议在方便时处理这些警告") return len(critical) == 0 def main(): """Main function""" parser = argparse.ArgumentParser( description='Data Validation and Integrity Check for XI Stock System', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 使用示例: data_validation.py all - 运行所有检查 data_validation.py db - 只检查数据库 data_validation.py files - 只检查数据文件 data_validation.py sync - 检查数据库与文件同步 data_validation.py timestamps - 检查时间戳一致性 """ ) parser.add_argument('check_type', choices=['all', 'db', 'files', 'sync', 'timestamps'], default='all', help='检查类型') args = parser.parse_args() all_issues = [] try: if args.check_type in ['all', 'db']: issues = check_database_integrity() all_issues.extend(issues) if args.check_type in ['all', 'files']: for frequency in ['day', 'week', 'month']: issues = check_data_files(frequency) all_issues.extend(issues) if args.check_type in ['all', 'sync']: issues = check_database_files_sync() all_issues.extend(issues) if args.check_type in ['all', 'timestamps']: issues = check_timestamp_consistency() all_issues.extend(issues) # Generate final report is_healthy = generate_report(all_issues) # Save report to file report_file = f"validation_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" with open(report_file, 'w', encoding='utf-8') as f: f.write(f"XI Stock System Validation Report\n") f.write(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"检查类型: {args.check_type}\n") f.write(f"发现问题: {len(all_issues)} 个\n\n") if all_issues: f.write("问题列表:\n") for i, issue in enumerate(all_issues, 1): f.write(f"{i}. {issue}\n") else: f.write("✅ 所有检查通过，系统健康状态良好。\n") print(f"\n📄 报告已保存: {report_file}") if not is_healthy: print(f"\n⚠️ 系统存在严重问题，建议运行修复脚本:") print(" python data_repair.py --fix-all") return 1 except KeyboardInterrupt: print("\n\n- 用户中断") return 1 except Exception as e: print(f"\n- 错误: {e}") import traceback traceback.print_exc() return 1 return 0 if __name__ == "__main__": exit(main()) FILE:scripts/day.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ XI Stock Daily Data Fetcher - Enhanced Version 羲股票监控系统 - 日线数据获取（增强版） This script fetches daily K-line data for stocks from Tencent Finance API. 本脚本从腾讯财经API获取股票的日K线数据。新增功能： day.py get [ac] - 获取股票数据 - ac: 股票代码（支持逗号分隔多个股票） - all: 获取所有未更新股票 - rand: 随机获取5只未更新股票 - 默认：获取5只未更新股票 """ import os import sys import sqlite3 import requests import json import random import argparse from datetime import datetime, timedelta # Fix encoding for Windows console # 修复Windows控制台编码问题 if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" DATA_DIR = "D:\\xistock\\day" BASE_URL = "https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get" DATA_POINTS = 800 FREQ = "day" DEFAULT_LIMIT = 5 def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- 数据库未找到: {DB_PATH}") print(" 请先运行 init_db.py 和 fetch_stocks.py!") exit(1) return sqlite3.connect(DB_PATH) def get_stocks_by_codes(stock_codes): """Get stocks by specific codes""" conn = get_db_connection() cursor = conn.cursor() # 处理逗号分隔的股票代码 codes = [code.strip() for code in stock_codes.split(',')] # 构建查询 placeholders = ','.join(['?'] * len(codes)) query = f""" SELECT code, name, market FROM stocks WHERE code IN ({placeholders}) AND status = 'active' ORDER BY code """ cursor.execute(query, codes) stocks = cursor.fetchall() conn.close() return stocks def get_all_stocks_to_update(limit=None): """Get all stocks that need to be updated (day_get is NULL or old)""" conn = get_db_connection() cursor = conn.cursor() # 获取当前时间 current_time = datetime.now() query = """ SELECT code, name, market FROM stocks WHERE status = 'active' AND (day_get IS NULL OR day_get < ?) ORDER BY day_get ASC NULLS FIRST """ cursor.execute(query, (current_time,)) stocks = cursor.fetchall() conn.close() # 应用限制 if limit and limit > 0: stocks = stocks[:limit] return stocks def get_random_stocks_to_update(limit=5): """Get random stocks that need to be updated""" conn = get_db_connection() cursor = conn.cursor() # 获取当前时间 current_time = datetime.now() query = """ SELECT code, name, market FROM stocks WHERE status = 'active' AND (day_get IS NULL OR day_get < ?) """ cursor.execute(query, (current_time,)) all_stocks = cursor.fetchall() conn.close() # 随机选择 if len(all_stocks) > limit: stocks = random.sample(all_stocks, limit) else: stocks = all_stocks return stocks def format_stock_code(code): """Format stock code for Tencent Finance API (sh600001, sz000001)""" if code.startswith('60'): return f"sh{code}" else: return f"sz{code}" def fetch_stock_data(stock_code, stock_name): """Fetch daily K-line data from Tencent Finance API""" try: tencent_code = format_stock_code(stock_code) params = f"{tencent_code},{FREQ},,,{DATA_POINTS},qfq" url = f"{BASE_URL}?_var=kline_dayqfq¶m={params}" response = requests.get(url, timeout=30) response.encoding = 'utf-8' if response.status_code != 200: print(f"- {stock_code} {stock_name}: HTTP {response.status_code}") return None # Response is in format: kline_dayqfq = {...}; content = response.text if '=' in content: json_str = content.split('=', 1)[1].rstrip(';') else: json_str = content data = json.loads(json_str) # Data structure: data -> tencent_code -> qfqday if 'data' not in data or tencent_code not in data['data']: print(f"- {stock_code} {stock_name}: No data found") return None stock_data = data['data'][tencent_code] if 'qfqday' not in stock_data: print(f"- {stock_code} {stock_name}: No qfqday found") return None return stock_data['qfqday'] except Exception as e: print(f"- {stock_code} {stock_name}: Error fetching - {str(e)}") return None def process_data(data): """ Process raw data, extract date, open, close, high, low and calculate change 返回格式列表: [date, open, close, high, low, change] """ processed = [] # Data format from Tencent: [date, open, close, high, low, volume, ...] for bar in data: if len(bar) >= 5: date = bar[0] open_p = float(bar[1]) close = float(bar[2]) high = float(bar[3]) low = float(bar[4]) processed.append([date, open_p, close, high, low]) # Calculate change (涨跌幅) for i in range(len(processed)): if i == 0: change = 0.0 else: prev_close = processed[i-1][2] if prev_close == 0: change = 0.0 else: change = ((processed[i][2] - prev_close) / prev_close) * 100 processed[i].append(round(change, 3)) return processed def save_to_file(stock_code, stock_name, data): """ Save processed data to file, one data point per line Format: date,open,close,high,low,change Filename: 名称_代码.txt """ # Create directory if not exists if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) # Sanitize filename - remove all invalid Windows filename characters # Invalid chars: \ / : * ? " < > | invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = stock_name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{stock_code}.txt" filepath = os.path.join(DATA_DIR, filename) # Write header with open(filepath, 'w', encoding='utf-8') as f: f.write("date,open,close,high,low,change_pct\n") for item in data: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) return filepath def update_database_last_fetch(stock_code): """Update the day_get timestamp in database to current date""" conn = get_db_connection() cursor = conn.cursor() current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') cursor.execute(""" UPDATE stocks SET day_get = ? WHERE code = ? """, (current_time, stock_code)) conn.commit() conn.close() def fetch_stocks(stocks): """Fetch data for given list of stocks""" total = len(stocks) if total == 0: print("没有需要更新的股票") return print("=" * 70) print("XI Stock Daily Data Fetcher") print("羲股票监控系统 - 日线数据获取") print("=" * 70) print(f"股票数量: {total}") print(f"数据目录: {DATA_DIR}") print("-" * 70) success = 0 failed = 0 for i, (code, name, market) in enumerate(stocks, 1): print(f"[{i}/{total}] 获取 {code} {name}...", end=' ') sys.stdout.flush() # Fetch data raw_data = fetch_stock_data(code, name) if raw_data is None: print("失败") failed += 1 continue # Process data bars = raw_data processed = process_data(bars) if not processed: print("无数据") failed += 1 continue # Save to file save_to_file(code, name, processed) # Update database update_database_last_fetch(code) print(f"成功 ({len(processed)} 个数据点)") success += 1 # Add delay to avoid hitting rate limit if i % 10 == 0: import time time.sleep(1) # Final summary print("-" * 70) print("任务完成！") print("我辛苦了！") print("-" * 70) print(f"总股票数: {total}") print(f"成功获取: {success}") print(f"失败: {failed}") print(f"数据保存到: {DATA_DIR}") print("-" * 70) print("数据库已更新最后获取时间戳") print("=" * 70) def fetch_all_stocks_original(): """Original function to fetch all stocks (backward compatibility)""" conn = get_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT code, name, market FROM stocks WHERE status = 'active' ORDER BY code """) stocks = cursor.fetchall() conn.close() fetch_stocks(stocks) def main(): """Main function with command line arguments""" parser = argparse.ArgumentParser( description='XI Stock Daily Data Fetcher - Enhanced Version', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 使用示例: 传统用法: day.py - 获取所有活跃股票数据增强用法: day.py get 000001 - 获取单只股票数据 day.py get 000001,000002 - 获取多只股票数据 day.py get all - 获取所有未更新股票 day.py get all --limit 10 - 获取10只未更新股票 day.py get rand - 随机获取5只未更新股票 day.py get rand --limit 3 - 随机获取3只未更新股票 day.py get - 获取5只未更新股票（默认） """ ) subparsers = parser.add_subparsers(dest='command', help='命令') # get command get_parser = subparsers.add_parser('get', help='获取股票数据') get_parser.add_argument('ac', nargs='?', default='', help='股票代码(all/rand/股票代码)') get_parser.add_argument('--limit', type=int, default=DEFAULT_LIMIT, help=f'获取股票数量限制 (默认: {DEFAULT_LIMIT})') args = parser.parse_args() try: if not args.command: # 传统用法：没有参数时获取所有股票 print("使用传统模式：获取所有活跃股票数据...") fetch_all_stocks_original() elif args.command == 'get': ac = args.ac.strip().lower() if args.ac else '' limit = args.limit if not ac: # 默认：获取5只未更新股票 print(f"未指定股票，获取 {limit} 只未更新股票...") stocks = get_all_stocks_to_update(limit) fetch_stocks(stocks) elif ac == 'all': # 获取所有未更新股票 if limit > 0: print(f"获取 {limit} 只未更新股票...") stocks = get_all_stocks_to_update(limit) else: print("获取所有未更新股票...") stocks = get_all_stocks_to_update() fetch_stocks(stocks) elif ac == 'rand': # 随机获取未更新股票 print(f"随机获取 {limit} 只未更新股票...") stocks = get_random_stocks_to_update(limit) fetch_stocks(stocks) else: # 获取指定股票代码 print(f"获取指定股票: {ac}") stocks = get_stocks_by_codes(ac) if not stocks: print(f"未找到股票代码: {ac}") return fetch_stocks(stocks) except KeyboardInterrupt: print("\n\n- 用户中断") exit(1) except Exception as e: print(f"\n- 错误: {e}") import traceback traceback.print_exc() exit(1) if __name__ == "__main__": main() FILE:scripts/day_original.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ XI Stock Daily Data Fetcher 羲股票监控系统 - 日线数据获取 This script fetches daily K-line data for all stocks from Tencent Finance API. 本脚本从腾讯财经API获取所有股票的日K线数据。 - Runs after 15:00 on every trading day - 在每个工作日15:00后执行 - Saves 800 historical data points per stock - 每只股票保存800个历史数据点 """ import os import sys import sqlite3 import requests import json from datetime import datetime # Fix encoding for Windows console # 修复Windows控制台编码问题 if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" DATA_DIR = "D:\\xistock\\day" BASE_URL = "https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get" DATA_POINTS = 800 FREQ = "day" def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- Database not found: {DB_PATH}") print(" Please run init_db.py and fetch_stocks.py first!") exit(1) return sqlite3.connect(DB_PATH) def get_stocks_to_fetch(): """Get list of stocks that need to be fetched (all active stocks)""" conn = get_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT code, name, market FROM stocks WHERE status = 'active' ORDER BY code """) stocks = cursor.fetchall() conn.close() return stocks def format_stock_code(code): """Format stock code for Tencent Finance API (sh600001, sz000001)""" if code.startswith('60'): return f"sh{code}" else: return f"sz{code}" def fetch_stock_data(stock_code, stock_name): """Fetch daily K-line data from Tencent Finance API""" try: tencent_code = format_stock_code(stock_code) params = f"{tencent_code},{FREQ},,,{DATA_POINTS},qfq" url = f"{BASE_URL}?_var=kline_dayqfq¶m={params}" response = requests.get(url, timeout=30) response.encoding = 'utf-8' if response.status_code != 200: print(f"- {stock_code} {stock_name}: HTTP {response.status_code}") return None # Response is in format: kline_dayqfq = {...}; content = response.text if '=' in content: json_str = content.split('=', 1)[1].rstrip(';') else: json_str = content data = json.loads(json_str) # Data structure: data -> tencent_code -> qfqday if 'data' not in data or tencent_code not in data['data']: print(f"- {stock_code} {stock_name}: No data found") return None stock_data = data['data'][tencent_code] if 'qfqday' not in stock_data: print(f"- {stock_code} {stock_name}: No qfqday found") return None return stock_data['qfqday'] except Exception as e: print(f"- {stock_code} {stock_name}: Error fetching - {str(e)}") return None def process_data(data): """ Process raw data, extract date, open, close, high, low and calculate change 返回格式列表: [date, open, close, high, low, change] """ processed = [] # Data format from Tencent: [date, open, close, high, low, volume, ...] for bar in data: if len(bar) >= 5: date = bar[0] open_p = float(bar[1]) close = float(bar[2]) high = float(bar[3]) low = float(bar[4]) processed.append([date, open_p, close, high, low]) # Calculate change (涨跌幅) for i in range(len(processed)): if i == 0: change = 0.0 else: prev_close = processed[i-1][2] if prev_close == 0: change = 0.0 else: change = ((processed[i][2] - prev_close) / prev_close) * 100 processed[i].append(round(change, 3)) return processed def save_to_file(stock_code, stock_name, data): """ Save processed data to file, one data point per line Format: date,open,close,high,low,change Filename: 名称_代码.txt """ # Create directory if not exists if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) # Sanitize filename - remove all invalid Windows filename characters # Invalid chars: \ / : * ? " < > | invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = stock_name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{stock_code}.txt" filepath = os.path.join(DATA_DIR, filename) # Write header with open(filepath, 'w', encoding='utf-8') as f: f.write("date,open,close,high,low,change_pct\n") for item in data: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) return filepath def update_database_last_fetch(stock_code): """Update the day_get timestamp in database to current date""" conn = get_db_connection() cursor = conn.cursor() current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') cursor.execute(""" UPDATE stocks SET day_get = ? WHERE code = ? """, (current_time, stock_code)) conn.commit() conn.close() def fetch_all_stocks(): """Fetch data for all stocks""" stocks = get_stocks_to_fetch() total = len(stocks) print("=" * 70) print("XI Stock Daily Data Fetcher") print("羲股票监控系统 - 日线数据获取") print("=" * 70) print(f"Total active stocks: {total}") print(f"Data directory: {DATA_DIR}") print("-" * 70) success = 0 failed = 0 for i, (code, name, market) in enumerate(stocks, 1): print(f"[{i}/{total}] Fetching {code} {name}...", end=' ') sys.stdout.flush() # Fetch data raw_data = fetch_stock_data(code, name) if raw_data is None: print("FAILED") failed += 1 continue # Process data bars = raw_data processed = process_data(bars) if not processed: print("NO DATA") failed += 1 continue # Save to file save_to_file(code, name, processed) # Update database update_database_last_fetch(code) print(f"OK ({len(processed)} points)") success += 1 # Add delay to avoid hitting rate limit if i % 10 == 0: import time time.sleep(1) # Final summary print("-" * 70) print("任务完成！") print("我辛苦了！") print("-" * 70) print(f"Total stocks: {total}") print(f"Successfully fetched: {success}") print(f"Failed: {failed}") print(f"Data saved to: {DATA_DIR}") print("-" * 70) print("Database updated with last fetch timestamps") print("=" * 70) if __name__ == "__main__": try: fetch_all_stocks() except KeyboardInterrupt: print("\n\n- Interrupted by user") exit(1) except Exception as e: print(f"\n- Error: {e}") import traceback traceback.print_exc() exit(1) FILE:scripts/day_parallel.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ A-Share-Get Daily Data Fetcher - Parallel Incremental Version A股数据获取 - 日线数据增量获取（并行版本） Usage: python scripts/day_parallel.py asc # 获取前一半（升序），每次2-3只 python scripts/day_parallel.py desc # 获取后一半（降序），每次2-3只 Run both in separate processes to speed up incremental updates! 在两个独立进程中同时运行，加速增量更新！ Incremental update logic: - Read existing file, get latest date - Only fetch data newer than latest date (add to file, don't overwrite) - If file doesn't exist, fetch full 800 points - Process max 100 stocks per run (incremental) / 800 (new file) - Update database timestamp after successful fetch """ import os import sys import sqlite3 import requests import json from datetime import datetime # Fix encoding for Windows console if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" DATA_DIR = "D:\\xistock\\day" BASE_URL = "https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get" DATA_POINTS_FULL = 800 # When file doesn't exist MAX_STOCKS_PER_RUN_INCREMENTAL = 100 # Max stocks to process per run when incremental update FREQ = "day" BATCH_SIZE = 3 # Process 2-3 stocks per batch to avoid long runtime def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- Database not found: {DB_PATH}") print(" Please run init_db.py and fetch_stocks.py first!") exit(1) return sqlite3.connect(DB_PATH) def get_latest_date_from_file(filepath): """ Read existing file and get the latest date Returns None if file doesn't exist or empty """ if not os.path.exists(filepath): return None try: with open(filepath, 'r', encoding='utf-8') as f: # Skip header next(f) latest_date = None # Read all lines to get the last date for line in f: line = line.strip() if not line: continue date = line.split(',')[0] latest_date = date return latest_date except Exception as e: print(f" Warning: Cannot read existing file: {e}") return None def get_stocks_to_fetch(order): """ Get list of stocks that need to be fetched (incremental) - Selects stocks that have never been fetched or haven't been fetched recently - order: 'asc' or 'desc' - split into two partitions - Returns up to MAX_STOCKS_PER_RUN_INCREMENTAL stocks """ conn = get_db_connection() cursor = conn.cursor() # Get all active stocks ordered by code cursor.execute(""" SELECT code, name, market, day_get FROM stocks WHERE status = 'active' ORDER BY code """) all_stocks = cursor.fetchall() total = len(all_stocks) half = total // 2 # Split into two partitions if order == 'asc': partition_stocks = all_stocks[:half] elif order == 'desc': all_stocks.reverse() partition_stocks = all_stocks[:half] else: print(f"- Invalid order: {order}. Use 'asc' or 'desc'") exit(1) # Filter stocks that need update (any stock that hasn't been fetched today) today = datetime.now().strftime('%Y-%m-%d') need_update = [] for stock in partition_stocks: code, name, market, day_get = stock # If never fetched or last fetched not today, needs update if day_get is None or not day_get.startswith(today): need_update.append((code, name, market)) # Process all need update today, large safety limit to prevent infinite run # MAX_STOCKS_PER_RUN_INCREMENTAL is original limit, now we allow 10x that if len(need_update) > MAX_STOCKS_PER_RUN_INCREMENTAL * 10: need_update = need_update[:MAX_STOCKS_PER_RUN_INCREMENTAL * 10] print(f"Partition {order}:") print(f" Total in partition: {len(partition_stocks)}") print(f" Need update today: {len(need_update)}") print(f" Will process all {len(need_update)} stocks (safety limit: {MAX_STOCKS_PER_RUN_INCREMENTAL * 10})") conn.close() return need_update def format_stock_code(code): """Format stock code for Tencent Finance API (sh600001, sz000001)""" if code.startswith('60'): return f"sh{code}" else: return f"sz{code}" def fetch_stock_data(stock_code, stock_name): """Fetch daily K-line data from Tencent Finance API""" try: tencent_code = format_stock_code(stock_code) params = f"{tencent_code},{FREQ},,,{DATA_POINTS_FULL},qfq" url = f"{BASE_URL}?_var=kline_dayqfq¶m={params}" response = requests.get(url, timeout=30) response.encoding = 'utf-8' if response.status_code != 200: print(f"- {stock_code} {stock_name}: HTTP {response.status_code}") return None # Response is in format: kline_dayqfq = {...}; content = response.text if '=' in content: json_str = content.split('=', 1)[1].rstrip(';') else: json_str = content data = json.loads(json_str) # Data structure: data -> tencent_code -> qfqday if 'data' not in data or tencent_code not in data['data']: print(f"- {stock_code} {stock_name}: No data found") return None stock_data = data['data'][tencent_code] if 'qfqday' not in stock_data: print(f"- {stock_code} {stock_name}: No qfqday found") return None return stock_data['qfqday'] except Exception as e: print(f"- {stock_code} {stock_name}: Error fetching - {str(e)}") return None def process_data(data): """ Process raw data, extract date, open, close, high, low and calculate change 返回格式列表: [date, open, close, high, low, change] """ processed = [] # Data format from Tencent: [date, open, close, high, low, volume, ...] for bar in data: if len(bar) >= 5: date = bar[0] open_p = float(bar[1]) close = float(bar[2]) high = float(bar[3]) low = float(bar[4]) processed.append([date, open_p, close, high, low]) # Calculate change (涨跌幅) for i in range(len(processed)): if i == 0: change = 0.0 else: prev_close = processed[i-1][2] if prev_close == 0: change = 0.0 else: change = ((processed[i][2] - prev_close) / prev_close) * 100 processed[i].append(round(change, 3)) return processed def get_data_points_to_fetch(processed_new, latest_date): """ Filter new data points that are after latest_date in existing file If latest_date is None, return all """ if latest_date is None: return processed_new, DATA_POINTS_FULL # Find all data points after latest_date new_data = [] for item in processed_new: date = item[0] if date > latest_date: new_data.append(item) return new_data, len(processed_new) def append_to_file(stock_code, stock_name, new_data, latest_date): """ Append new data points to existing file, don't overwrite If file doesn't exist, create new with header Format: date,open,close,high,low,change Filename: 名称_代码.txt """ # Create directory if not exists if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) # Sanitize filename - remove all invalid Windows filename characters invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = stock_name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{stock_code}.txt" filepath = os.path.join(DATA_DIR, filename) # Filter new data (only data after latest_date) data_to_write, total_fetched = get_data_points_to_fetch(new_data, latest_date) if not data_to_write: # No new data to add return filepath, 0 if latest_date is None: # Create new file with open(filepath, 'w', encoding='utf-8') as f: f.write("date,open,close,high,low,change_pct\n") for item in data_to_write: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) else: # Append to existing file with open(filepath, 'a', encoding='utf-8') as f: for item in data_to_write: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) return filepath, len(data_to_write) def update_database_last_fetch(stock_code): """Update the day_get timestamp in database to current date""" conn = get_db_connection() cursor = conn.cursor() current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') cursor.execute(""" UPDATE stocks SET day_get = ? WHERE code = ? """, (current_time, stock_code)) conn.commit() conn.close() def fetch_all_stocks(order): """Fetch data incrementally for stocks in this partition - process {BATCH_SIZE} per run""" stocks = get_stocks_to_fetch(order) total_need_update = len(stocks) print("=" * 70) print(f"A-Share-Get Daily Data Fetcher - Incremental Parallel ({order})") print("A股数据获取 - 日线增量更新（并行）") print("=" * 70) print(f"Data directory: {DATA_DIR}") print(f"Processing {BATCH_SIZE} stocks per run (max {MAX_STOCKS_PER_RUN_INCREMENTAL} total)") print("-" * 70) success_total = 0 failed_total = 0 new_points_total = 0 # Process in batches of BATCH_SIZE from itertools import islice def batch_generator(iterable, batch_size): iterator = iter(iterable) for first in iterator: yield [first] + list(islice(iterator, batch_size - 1)) for batch_idx, batch in enumerate(batch_generator(stocks, BATCH_SIZE), 1): print(f"\n>> Batch {batch_idx}, processing {len(batch)} stocks:") for i, (code, name, market) in enumerate(batch, 1): global_i = (batch_idx - 1) * BATCH_SIZE + i print(f" [{global_i}/{total_need_update}] {code} {name}: ", end=' ') sys.stdout.flush() # Get existing file's latest date invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{code}.txt" filepath = os.path.join(DATA_DIR, filename) latest_date = get_latest_date_from_file(filepath) if latest_date: print(f"latest={latest_date}, fetching new data...", end=' ') else: print("new file, fetching full data...", end=' ') sys.stdout.flush() # Fetch data raw_data = fetch_stock_data(code, name) if raw_data is None: print("FAILED") failed_total += 1 continue # Process data processed = process_data(raw_data) if not processed: print("NO DATA") failed_total += 1 continue # Append new data to file (only add missing) filepath, new_count = append_to_file(code, name, processed, latest_date) if new_count == 0: print("no new data") # Still update database to mark as updated today update_database_last_fetch(code) continue # Update database update_database_last_fetch(code) print(f"OK added {new_count} new points") success_total += 1 new_points_total += new_count # Add delay to avoid hitting rate limit import time time.sleep(1.5) # Final summary print("-" * 70) print("增量更新完成！") print("-" * 70) print(f"Partition ({order}):") print(f" - Need update: {total_need_update}") print(f" - Successfully processed: {success_total}") print(f" - Failed: {failed_total}") print(f" - New data points added: {new_points_total}") print(f"Data directory: {DATA_DIR}") print("-" * 70) print("Database updated with last fetch timestamps") print("=" * 70) if __name__ == "__main__": if len(sys.argv) != 2: print("Usage:") print(" python scripts/day_parallel.py asc # First half (ascending)") print(" python scripts/day_parallel.py desc # Second half (descending)") print("\nRun both in separate processes for parallel fetching!") exit(1) order = sys.argv[1].lower() try: fetch_all_stocks(order) except KeyboardInterrupt: print("\n\n- Interrupted by user") exit(1) except Exception as e: print(f"\n- Error: {e}") import traceback traceback.print_exc() exit(1) FILE:scripts/db_reset.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Database Reset and Fetch Tool for XI Stock System 羲股票监控系统数据库重置与数据获取工具 This script resets specific timestamp fields and fetches data for stocks. 本脚本重置数据库中的特定时间戳字段并获取股票数据。 """ import os import sys import sqlite3 import argparse import requests import json import random from datetime import datetime, timedelta # Fix encoding for Windows console if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" BASE_URL = "https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get" DATA_POINTS = 800 def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- 数据库未找到: {DB_PATH}") print(" 请先运行 init_db.py 和 fetch_stocks.py!") exit(1) return sqlite3.connect(DB_PATH) # ==================== RESET FUNCTIONS ==================== def reset_day_get(): """Reset day_get field for all stocks""" conn = get_db_connection() cursor = conn.cursor() try: cursor.execute(""" UPDATE stocks SET day_get = NULL """) affected_rows = cursor.rowcount conn.commit() print("=" * 70) print("重置 day_get 字段") print("=" * 70) print(f"已重置 {affected_rows} 只股票的 day_get 字段为 NULL") print("所有股票现在都标记为需要更新日线数据") print("=" * 70) except Exception as e: print(f"- 重置失败: {e}") conn.rollback() finally: conn.close() def reset_week_get(): """Reset week_get field for all stocks""" conn = get_db_connection() cursor = conn.cursor() try: cursor.execute(""" UPDATE stocks SET week_get = NULL """) affected_rows = cursor.rowcount conn.commit() print("=" * 70) print("重置 week_get 字段") print("=" * 70) print(f"已重置 {affected_rows} 只股票的 week_get 字段为 NULL") print("所有股票现在都标记为需要更新周线数据") print("=" * 70) except Exception as e: print(f"- 重置失败: {e}") conn.rollback() finally: conn.close() def reset_month_get(): """Reset month_get field for all stocks""" conn = get_db_connection() cursor = conn.cursor() try: cursor.execute(""" UPDATE stocks SET month_get = NULL """) affected_rows = cursor.rowcount conn.commit() print("=" * 70) print("重置 month_get 字段") print("=" * 70) print(f"已重置 {affected_rows} 只股票的 month_get 字段为 NULL") print("所有股票现在都标记为需要更新月线数据") print("=" * 70) except Exception as e: print(f"- 重置失败: {e}") conn.rollback() finally: conn.close() def reset_all(): """Reset all timestamp fields for all stocks""" conn = get_db_connection() cursor = conn.cursor() try: cursor.execute(""" UPDATE stocks SET day_get = NULL, week_get = NULL, month_get = NULL """) affected_rows = cursor.rowcount conn.commit() print("=" * 70) print("重置所有时间戳字段") print("=" * 70) print(f"已重置 {affected_rows} 只股票的所有时间戳字段") print("重置字段: day_get, week_get, month_get") print("所有股票现在都标记为需要更新全部数据") print("=" * 70) except Exception as e: print(f"- 重置失败: {e}") conn.rollback() finally: conn.close() def show_status(): """Show current database status""" conn = get_db_connection() cursor = conn.cursor() try: # Total stocks cursor.execute("SELECT COUNT(*) FROM stocks") total = cursor.fetchone()[0] # Stocks with day_get cursor.execute("SELECT COUNT(*) FROM stocks WHERE day_get IS NOT NULL") day_updated = cursor.fetchone()[0] # Stocks with week_get cursor.execute("SELECT COUNT(*) FROM stocks WHERE week_get IS NOT NULL") week_updated = cursor.fetchone()[0] # Stocks with month_get cursor.execute("SELECT COUNT(*) FROM stocks WHERE month_get IS NOT NULL") month_updated = cursor.fetchone()[0] print("=" * 70) print("数据库状态报告") print("=" * 70) print(f"总股票数: {total}") print(f"日线数据已更新: {day_updated} ({day_updated/total*100:.1f}%)") print(f"周线数据已更新: {week_updated} ({week_updated/total*100:.1f}%)") print(f"月线数据已更新: {month_updated} ({month_updated/total*100:.1f}%)") print("=" * 70) except Exception as e: print(f"- 查询失败: {e}") finally: conn.close() # ==================== FETCH FUNCTIONS ==================== def get_stocks_by_codes(stock_codes): """Get stocks by specific codes""" conn = get_db_connection() cursor = conn.cursor() # 处理逗号分隔的股票代码 codes = [code.strip() for code in stock_codes.split(',')] # 构建查询 placeholders = ','.join(['?'] * len(codes)) query = f""" SELECT code, name, market FROM stocks WHERE code IN ({placeholders}) AND status = 'active' ORDER BY code """ cursor.execute(query, codes) stocks = cursor.fetchall() conn.close() return stocks def get_all_stocks_to_update(frequency, limit=None): """Get all stocks that need to be updated for specific frequency""" conn = get_db_connection() cursor = conn.cursor() # 获取当前时间 current_time = datetime.now() # 根据频率确定字段名 if frequency == 'day': timestamp_field = 'day_get' elif frequency == 'week': timestamp_field = 'week_get' elif frequency == 'month': timestamp_field = 'month_get' else: timestamp_field = 'day_get' query = f""" SELECT code, name, market FROM stocks WHERE status = 'active' AND ({timestamp_field} IS NULL OR {timestamp_field} < ?) ORDER BY {timestamp_field} ASC NULLS FIRST """ cursor.execute(query, (current_time,)) stocks = cursor.fetchall() conn.close() # 应用限制 if limit and limit > 0: stocks = stocks[:limit] return stocks def get_random_stocks_to_update(frequency, limit=5): """Get random stocks that need to be updated for specific frequency""" conn = get_db_connection() cursor = conn.cursor() # 获取当前时间 current_time = datetime.now() # 根据频率确定字段名 if frequency == 'day': timestamp_field = 'day_get' elif frequency == 'week': timestamp_field = 'week_get' elif frequency == 'month': timestamp_field = 'month_get' else: timestamp_field = 'day_get' query = f""" SELECT code, name, market FROM stocks WHERE status = 'active' AND ({timestamp_field} IS NULL OR {timestamp_field} < ?) """ cursor.execute(query, (current_time,)) all_stocks = cursor.fetchall() conn.close() # 随机选择 if len(all_stocks) > limit: stocks = random.sample(all_stocks, limit) else: stocks = all_stocks return stocks def format_stock_code(code): """Format stock code for Tencent Finance API (sh600001, sz000001)""" if code.startswith('60'): return f"sh{code}" else: return f"sz{code}" def fetch_stock_data(stock_code, stock_name, frequency): """Fetch K-line data from Tencent Finance API""" try: tencent_code = format_stock_code(stock_code) params = f"{tencent_code},{frequency},,,{DATA_POINTS},qfq" url = f"{BASE_URL}?_var=kline_{frequency}qfq¶m={params}" response = requests.get(url, timeout=30) response.encoding = 'utf-8' if response.status_code != 200: print(f"- {stock_code} {stock_name}: HTTP {response.status_code}") return None # Response is in format: kline_{frequency}qfq = {...}; content = response.text if '=' in content: json_str = content.split('=', 1)[1].rstrip(';') else: json_str = content data = json.loads(json_str) # Data structure: data -> tencent_code -> qfq{frequency} if 'data' not in data or tencent_code not in data['data']: print(f"- {stock_code} {stock_name}: No data found") return None stock_data = data['data'][tencent_code] data_key = f"qfq{frequency}" if data_key not in stock_data: print(f"- {stock_code} {stock_name}: No {data_key} found") return None return stock_data[data_key] except Exception as e: print(f"- {stock_code} {stock_name}: Error fetching - {str(e)}") return None def process_data(data): """ Process raw data, extract date, open, close, high, low and calculate change 返回格式列表: [date, open, close, high, low, change] """ processed = [] # Data format from Tencent: [date, open, close, high, low, volume, ...] for bar in data: if len(bar) >= 5: date = bar[0] open_p = float(bar[1]) close = float(bar[2]) high = float(bar[3]) low = float(bar[4]) processed.append([date, open_p, close, high, low]) # Calculate change (涨跌幅) for i in range(len(processed)): if i == 0: change = 0.0 else: prev_close = processed[i-1][2] if prev_close == 0: change = 0.0 else: change = ((processed[i][2] - prev_close) / prev_close) * 100 processed[i].append(round(change, 3)) return processed def save_to_file(stock_code, stock_name, data, frequency): """ Save processed data to file, one data point per line Format: date,open,close,high,low,change Filename: 名称_代码.txt """ # 确定数据目录 if frequency == 'day': data_dir = "D:\\xistock\\day" elif frequency == 'week': data_dir = "D:\\xistock\\week" elif frequency == 'month': data_dir = "D:\\xistock\\month" else: data_dir = "D:\\xistock\\unknown" # Create directory if not exists if not os.path.exists(data_dir): os.makedirs(data_dir) # Sanitize filename - remove all invalid Windows filename characters # Invalid chars: \ / : * ? " < > | invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = stock_name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{stock_code}.txt" filepath = os.path.join(data_dir, filename) # Write header with open(filepath, 'w', encoding='utf-8') as f: f.write("date,open,close,high,low,change_pct\n") for item in data: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) return filepath def update_database_last_fetch(stock_code, frequency): """Update the timestamp field in database to current date""" conn = get_db_connection() cursor = conn.cursor() current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 根据频率确定字段名 if frequency == 'day': timestamp_field = 'day_get' elif frequency == 'week': timestamp_field = 'week_get' elif frequency == 'month': timestamp_field = 'month_get' else: timestamp_field = 'day_get' cursor.execute(f""" UPDATE stocks SET {timestamp_field} = ? WHERE code = ? """, (current_time, stock_code)) conn.commit() conn.close() def fetch_stocks_data(stocks, frequency, data_type): """Fetch data for given list of stocks""" total = len(stocks) if total == 0: print("没有需要更新的股票") return print("=" * 70) print(f"XI Stock {data_type.capitalize()} Data Fetcher") print(f"羲股票监控系统 - {data_type}数据获取") print("=" * 70) print(f"股票数量: {total}") print(f"数据频率: {frequency}") print("-" * 70) success = 0 failed = 0 for i, (code, name, market) in enumerate(stocks, 1): print(f"[{i}/{total}] 获取 {code} {name}...", end=' ') sys.stdout.flush() # Fetch data raw_data = fetch_stock_data(code, name, frequency) if raw_data is None: print("失败") failed += 1 continue # Process data bars = raw_data processed = process_data(bars) if not processed: print("无数据") failed += 1 continue # Save to file save_to_file(code, name, processed, frequency) # Update database update_database_last_fetch(code, frequency) print(f"成功 ({len(processed)} 个数据点)") success += 1 # Add delay to avoid hitting rate limit if i % 10 == 0: import time time.sleep(1) # Final summary print("-" * 70) print("任务完成！") print("-" * 70) print(f"总股票数: {total}") print(f"成功获取: {success}") print(f"失败: {failed}") print("-" * 70) print("数据库已更新最后获取时间戳") print("=" * 70) def fetch_day_data(action, limit=None): """Fetch day data based on action""" if action == 'all': print("获取所有未更新日线股票...") stocks = get_all_stocks_to_update('day', limit) elif action == 'rand': limit = limit or 5 print(f"随机获取 {limit} 只未更新日线股票...") stocks = get_random_stocks_to_update('day', limit) else: print(f"获取指定日线股票: {action}") stocks = get_stocks_by_codes(action) if not stocks: print(f"未找到股票代码: {action}") return fetch_stocks_data(stocks, 'day', '日线') def fetch_week_data(action, limit=None): """Fetch week data based on action""" if action == 'all': print("获取所有未更新周线股票...") stocks = get_all_stocks_to_update('week', limit) elif action == 'rand': limit = limit or 5 print(f"随机获取 {limit} 只未更新周线股票...") stocks = get_random_stocks_to_update('week', limit) else: print(f"获取指定周线股票: {action}") stocks = get_stocks_by_codes(action) if not stocks: print(f"未找到股票代码: {action}") return fetch_stocks_data(stocks, 'week', '周线') def fetch_month_data(action, limit=None): """Fetch month data based on action""" if action == 'all': print("获取所有未更新月线股票...") stocks = get_all_stocks_to_update('month', limit) elif action == 'rand': limit = limit or 5 print(f"随机获取 {limit} 只未更新月线股票...") stocks = get_random_stocks_to_update('month', limit) else: print(f"获取指定月线股票: {action}") stocks = get_stocks_by_codes(action) if not stocks: print(f"未找到股票代码: {action}") return fetch_stocks_data(stocks, 'month', '月线') def main(): """Main function""" parser = argparse.ArgumentParser( description='Database Reset and Fetch Tool for XI Stock System', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 使用示例: 重置功能: db_reset.py reset day - 重置所有股票的day_get字段 db_reset.py reset week - 重置所有股票的week_get字段 db_reset.py reset month - 重置所有股票的month_get字段 db_reset.py reset all - 重置所有时间戳字段 db_reset.py reset status - 显示数据库状态获取功能: db_reset.py fetch day 000001 - 获取单只日线股票数据 db_reset.py fetch day 000001,000002 - 获取多只日线股票数据 db_reset.py fetch day all - 获取所有未更新日线股票 db_reset.py fetch day all --limit 10 - 获取10只未更新日线股票 db_reset.py fetch day rand - 随机获取5只未更新日线股票 db_reset.py fetch day rand --limit 3 - 随机获取3只未更新日线股票 db_reset.py fetch week ... - 周线数据获取 db_reset.py fetch month ... - 月线数据获取注意事项: 1. 重置后会标记所有股票为需要更新状态 2. 请谨慎操作，建议在非交易时段进行 3. 重置后运行相应的数据获取脚本更新数据 """ ) subparsers = parser.add_subparsers(dest='command', help='命令', required=True) # reset command reset_parser = subparsers.add_parser('reset', help='重置数据库字段') reset_parser.add_argument('action', choices=['day', 'week', 'month', 'all', 'status'], help='执行的操作') # fetch command fetch_parser = subparsers.add_parser('fetch', help='获取股票数据') fetch_parser.add_argument('frequency', choices=['day', 'week', 'month'], help='数据频率') fetch_parser.add_argument('action', help='股票代码(all/rand/股票代码)') fetch_parser.add_argument('--limit', type=int, help='获取股票数量限制') args = parser.parse_args() try: if args.command == 'reset': if args.action == 'day': reset_day_get() elif args.action == 'week': reset_week_get() elif args.action == 'month': reset_month_get() elif args.action == 'all': reset_all() elif args.action == 'status': show_status() elif args.command == 'fetch': if args.frequency == 'day': fetch_day_data(args.action, args.limit) elif args.frequency == 'week': fetch_week_data(args.action, args.limit) elif args.frequency == 'month': fetch_month_data(args.action, args.limit) except KeyboardInterrupt: print("\n\n- 用户中断") exit(1) except Exception as e: print(f"\n- 错误: {e}") import traceback traceback.print_exc() exit(1) if __name__ == "__main__": main() FILE:scripts/fetch_stocks.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ XI Stock Fetcher - Stock List Acquisition 羲股票监控系统 - 股票列表获取 This script fetches tradable stocks from A-share markets (60*, 30*, 00*) and stores them in the database, excluding delisted and pre-IPO stocks. 本脚本从A股市场获取可交易股票（60*、30*、00*），并排除退市和未上市股票。 """ import sqlite3 import os import sys import time from datetime import datetime try: import akshare as ak import pandas as pd except ImportError: print("- Required packages not installed!") print(" Please run: pip install akshare pandas") sys.exit(1) # Database path DB_PATH = "D:\\xistock\\stock.db" def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"✗ Database not found: {DB_PATH}") print(" Please run init_db.py first!") sys.exit(1) return sqlite3.connect(DB_PATH) def fetch_stock_list(): """ Fetch tradable A-share stocks from Shanghai and Shenzhen markets 获取沪市和深市可交易的A股股票 """ print("=" * 60) print("Fetching A-Share Stock List") print("获取A股股票列表") print("=" * 60) try: # Get Shanghai A-shares (60*) print("\n[1/3] Fetching Shanghai A-shares (60*)...") sh_stocks = ak.stock_info_sh_name_code() sh_stocks = sh_stocks[sh_stocks['SECUCODE'].str.startswith('60')] print(f" + Found {len(sh_stocks)} Shanghai stocks") # Get Shenzhen A-shares (00* and 30*) print("\n[2/3] Fetching Shenzhen A-shares (00* and 30*)...") sz_stocks = ak.stock_info_sz_name_code() sz_stocks_00 = sz_stocks[sz_stocks['A股代码'].astype(str).str.startswith('00')] sz_stocks_30 = sz_stocks[sz_stocks['A股代码'].astype(str).str.startswith('30')] print(f" + Found {len(sz_stocks_00)} Shenzhen main board stocks (00*)") print(f" + Found {len(sz_stocks_30)} Shenzhen ChiNext stocks (30*)") except Exception as e: print(f"- Error fetching from akshare: {e}") print(" Trying alternative method...") return fetch_stock_list_alternative() # Combine all stocks all_stocks = [] # Process Shanghai stocks for _, stock in sh_stocks.iterrows(): code = str(stock['SECUCODE']).split('.')[0] name = stock['SECURITY_ABBR'] all_stocks.append({ 'code': code, 'name': name, 'market': '60' }) # Process Shenzhen 00* stocks for _, stock in sz_stocks_00.iterrows(): code = str(stock['A股代码']).zfill(6) name = stock['A股简称'] all_stocks.append({ 'code': code, 'name': name, 'market': '00' }) # Process Shenzhen 30* stocks for _, stock in sz_stocks_30.iterrows(): code = str(stock['A股代码']).zfill(6) name = stock['A股简称'] all_stocks.append({ 'code': code, 'name': name, 'market': '30' }) print(f"\n[3/3] Total stocks collected: {len(all_stocks)}") return all_stocks def fetch_stock_list_alternative(): """ Alternative method to fetch stock list 备用方法获取股票列表 """ print("\nUsing alternative data source...") try: # Get all A-share stocks all_stocks_df = ak.stock_info_a_code_name() all_stocks = [] for _, stock in all_stocks_df.iterrows(): code = str(stock['code']).zfill(6) name = stock['name'] # Determine market type if code.startswith('60'): market = '60' elif code.startswith('00'): market = '00' elif code.startswith('30'): market = '30' else: continue # Skip other markets all_stocks.append({ 'code': code, 'name': name, 'market': market }) print(f"+ Found {len(all_stocks)} stocks") return all_stocks except Exception as e: print(f"- Alternative method also failed: {e}") return [] def save_to_database(stocks): """ Save stock list to database 保存股票列表到数据库 """ if not stocks: print("- No stocks to save!") return 0 conn = get_db_connection() cursor = conn.cursor() # Count existing stocks cursor.execute("SELECT COUNT(*) FROM stocks") existing_count = cursor.fetchone()[0] print(f"\nFound {existing_count} existing stocks in database") # Insert or update stocks inserted = 0 updated = 0 for stock in stocks: try: # Check if stock already exists cursor.execute("SELECT code FROM stocks WHERE code = ?", (stock['code'],)) if cursor.fetchone(): # Update existing record cursor.execute(""" UPDATE stocks SET name = ?, market = ?, status = 'active' WHERE code = ? """, (stock['name'], stock['market'], stock['code'])) updated += 1 else: # Insert new record cursor.execute(""" INSERT INTO stocks (code, name, market, status) VALUES (?, ?, ?, 'active') """, (stock['code'], stock['name'], stock['market'])) inserted += 1 except sqlite3.Error as e: print(f"- Error saving stock {stock['code']}: {e}") continue conn.commit() conn.close() print(f"\n+ Database update completed:") print(f" - Newly inserted: {inserted}") print(f" - Updated: {updated}") print(f" - Total in database: {existing_count + inserted}") return existing_count + inserted def show_stock_stats(): """ Show statistics of stocks in database 显示数据库中的股票统计信息 """ conn = get_db_connection() cursor = conn.cursor() print("\n" + "=" * 60) print("Current Stock Statistics") print("当前股票统计信息") print("=" * 60) # Total count cursor.execute("SELECT COUNT(*) FROM stocks WHERE status = 'active'") total = cursor.fetchone()[0] print(f"\nTotal active stocks: {total}") # Count by market cursor.execute(""" SELECT market, COUNT(*) as count FROM stocks WHERE status = 'active' GROUP BY market ORDER BY market """) print("\nBy market:") for market, count in cursor.fetchall(): market_name = { '60': 'Shanghai (沪市)', '00': 'Shenzhen Main (深市主板)', '30': 'Shenzhen ChiNext (创业板)' }.get(market, market) print(f" - {market_name}: {count}") # Show sample stocks print("\nSample stocks:") cursor.execute(""" SELECT code, name, market FROM stocks WHERE status = 'active' ORDER BY RANDOM() LIMIT 5 """) for code, name, market in cursor.fetchall(): print(f" - {code} ({market}): {name}") conn.close() if __name__ == "__main__": print("XI Stock Fetcher") print("羲股票监控系统 - 股票列表获取") print("=" * 60) try: # Check if database exists if not os.path.exists(DB_PATH): print(f"- Database not found: {DB_PATH}") print(" Please run init_db.py first!") sys.exit(1) # Fetch stock list stocks = fetch_stock_list() if not stocks: print("- No stocks fetched!") sys.exit(1) # Save to database total = save_to_database(stocks) # Show statistics show_stock_stats() print("\n" + "=" * 60) print("任务完成！") print("我辛苦了！") print("-" * 60) print(f"成功获取并保存了 {total} 只可交易A股股票") print("数据已存入数据库: D:\\xistock\\stock.db") print("=" * 60) print("+ Stock list acquisition completed successfully!") print("=" * 60) except Exception as e: print(f"\n- Error: {e}") import traceback traceback.print_exc() sys.exit(1) FILE:scripts/init_db.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ XI Stock Database Initialization Script 羲股票监控系统数据库初始化脚本 This script initializes the SQLite database and creates the stocks table. 本脚本初始化 SQLite 数据库并创建股票列表表。 """ import sqlite3 import os from datetime import datetime # Database path DB_DIR = "D:\\xistock" DB_PATH = os.path.join(DB_DIR, "stock.db") def init_database(): """Initialize database and create tables""" # Create directory if not exists if not os.path.exists(DB_DIR): os.makedirs(DB_DIR) print(f"+ Created directory: {DB_DIR}") # Connect to database conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() # Create stocks table cursor.execute(''' CREATE TABLE IF NOT EXISTS stocks ( code TEXT PRIMARY KEY, name TEXT NOT NULL, market TEXT NOT NULL, day_get TIMESTAMP, week_get TIMESTAMP, month_get TIMESTAMP, status TEXT DEFAULT 'active', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Create index for faster queries cursor.execute(''' CREATE INDEX IF NOT EXISTS idx_market ON stocks(market) ''') cursor.execute(''' CREATE INDEX IF NOT EXISTS idx_status ON stocks(status) ''') conn.commit() conn.close() print(f"+ Database initialized successfully: {DB_PATH}") print("+ Table 'stocks' created with columns:") print(" - code: Stock code (PRIMARY KEY)") print(" - name: Stock name") print(" - market: Market type (60/30/00)") print(" - day_get: Last daily data fetch time") print(" - week_get: Last weekly data fetch time") print(" - month_get: Last monthly data fetch time") print(" - status: Stock status (active/delisted)") print(" - created_at: Record creation time") if __name__ == "__main__": print("=" * 60) print("XI Stock Database Initialization") print("羲股票监控系统数据库初始化") print("=" * 60) print(f"Database path: {DB_PATH}") print("-" * 60) try: init_database() print("-" * 60) print("任务完成！") print("我辛苦了！") print("-" * 60) print("+ Database initialized at: D:\\xistock\\stock.db") print("+ Initialization completed successfully!") except Exception as e: print(f"- Error: {e}") exit(1) FILE:scripts/month.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ XI Stock Monthly Data Fetcher - Enhanced Version 羲股票监控系统 - 月线数据获取（增强版） This script fetches monthly K-line data for stocks from Tencent Finance API. 本脚本从腾讯财经API获取股票的月K线数据。新增功能： month.py get [ac] - 获取股票数据 - ac: 股票代码（支持逗号分隔多个股票） - all: 获取所有未更新股票 - rand: 随机获取5只未更新股票 - 默认：获取5只未更新股票 """ import os import sys import sqlite3 import requests import json import random import argparse from datetime import datetime, timedelta # Fix encoding for Windows console # 修复Windows控制台编码问题 if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" DATA_DIR = "D:\\xistock\\month" BASE_URL = "https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get" DATA_POINTS = 800 FREQ = "month" DEFAULT_LIMIT = 5 TIMESTAMP_FIELD = "month_get" DATA_TYPE = "月线" def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- 数据库未找到: {DB_PATH}") print(" 请先运行 init_db.py 和 fetch_stocks.py!") exit(1) return sqlite3.connect(DB_PATH) def get_stocks_by_codes(stock_codes): """Get stocks by specific codes""" conn = get_db_connection() cursor = conn.cursor() # 处理逗号分隔的股票代码 codes = [code.strip() for code in stock_codes.split(',')] # 构建查询 placeholders = ','.join(['?'] * len(codes)) query = f""" SELECT code, name, market FROM stocks WHERE code IN ({placeholders}) AND status = 'active' ORDER BY code """ cursor.execute(query, codes) stocks = cursor.fetchall() conn.close() return stocks def get_all_stocks_to_update(limit=None): """Get all stocks that need to be updated (month_get is NULL or old)""" conn = get_db_connection() cursor = conn.cursor() # 获取当前时间 current_time = datetime.now() query = f""" SELECT code, name, market FROM stocks WHERE status = 'active' AND ({TIMESTAMP_FIELD} IS NULL OR {TIMESTAMP_FIELD} < ?) ORDER BY {TIMESTAMP_FIELD} ASC NULLS FIRST """ cursor.execute(query, (current_time,)) stocks = cursor.fetchall() conn.close() # 应用限制 if limit and limit > 0: stocks = stocks[:limit] return stocks def get_random_stocks_to_update(limit=5): """Get random stocks that need to be updated""" conn = get_db_connection() cursor = conn.cursor() # 获取当前时间 current_time = datetime.now() query = f""" SELECT code, name, market FROM stocks WHERE status = 'active' AND ({TIMESTAMP_FIELD} IS NULL OR {TIMESTAMP_FIELD} < ?) """ cursor.execute(query, (current_time,)) all_stocks = cursor.fetchall() conn.close() # 随机选择 if len(all_stocks) > limit: stocks = random.sample(all_stocks, limit) else: stocks = all_stocks return stocks def format_stock_code(code): """Format stock code for Tencent Finance API (sh600001, sz000001)""" if code.startswith('60'): return f"sh{code}" else: return f"sz{code}" def fetch_stock_data(stock_code, stock_name): """Fetch monthly K-line data from Tencent Finance API""" try: tencent_code = format_stock_code(stock_code) params = f"{tencent_code},{FREQ},,,{DATA_POINTS},qfq" url = f"{BASE_URL}?_var=kline_{FREQ}qfq¶m={params}" response = requests.get(url, timeout=30) response.encoding = 'utf-8' if response.status_code != 200: print(f"- {stock_code} {stock_name}: HTTP {response.status_code}") return None # Response is in format: kline_monthqfq = {...}; content = response.text if '=' in content: json_str = content.split('=', 1)[1].rstrip(';') else: json_str = content data = json.loads(json_str) # Data structure: data -> tencent_code -> qfqmonth if 'data' not in data or tencent_code not in data['data']: print(f"- {stock_code} {stock_name}: No data found") return None stock_data = data['data'][tencent_code] data_key = f"qfq{FREQ}" if data_key not in stock_data: print(f"- {stock_code} {stock_name}: No {data_key} found") return None return stock_data[data_key] except Exception as e: print(f"- {stock_code} {stock_name}: Error fetching - {str(e)}") return None def process_data(data): """ Process raw data, extract date, open, close, high, low and calculate change 返回格式列表: [date, open, close, high, low, change] """ processed = [] # Data format from Tencent: [date, open, close, high, low, volume, ...] for bar in data: if len(bar) >= 5: date = bar[0] open_p = float(bar[1]) close = float(bar[2]) high = float(bar[3]) low = float(bar[4]) processed.append([date, open_p, close, high, low]) # Calculate change (涨跌幅) for i in range(len(processed)): if i == 0: change = 0.0 else: prev_close = processed[i-1][2] if prev_close == 0: change = 0.0 else: change = ((processed[i][2] - prev_close) / prev_close) * 100 processed[i].append(round(change, 3)) return processed def save_to_file(stock_code, stock_name, data): """ Save processed data to file, one data point per line Format: date,open,close,high,low,change Filename: 名称_代码.txt """ # Create directory if not exists if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) # Sanitize filename - remove all invalid Windows filename characters # Invalid chars: \ / : * ? " < > | invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = stock_name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{stock_code}.txt" filepath = os.path.join(DATA_DIR, filename) # Write header with open(filepath, 'w', encoding='utf-8') as f: f.write("date,open,close,high,low,change_pct\n") for item in data: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) return filepath def update_database_last_fetch(stock_code): """Update the month_get timestamp in database to current date""" conn = get_db_connection() cursor = conn.cursor() current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') cursor.execute(f""" UPDATE stocks SET {TIMESTAMP_FIELD} = ? WHERE code = ? """, (current_time, stock_code)) conn.commit() conn.close() def fetch_stocks(stocks): """Fetch data for given list of stocks""" total = len(stocks) if total == 0: print("没有需要更新的股票") return print("=" * 70) print(f"XI Stock {DATA_TYPE.capitalize()} Data Fetcher") print(f"羲股票监控系统 - {DATA_TYPE}数据获取") print("=" * 70) print(f"股票数量: {total}") print(f"数据目录: {DATA_DIR}") print("-" * 70) success = 0 failed = 0 for i, (code, name, market) in enumerate(stocks, 1): print(f"[{i}/{total}] 获取 {code} {name}...", end=' ') sys.stdout.flush() # Fetch data raw_data = fetch_stock_data(code, name) if raw_data is None: print("失败") failed += 1 continue # Process data bars = raw_data processed = process_data(bars) if not processed: print("无数据") failed += 1 continue # Save to file save_to_file(code, name, processed) # Update database update_database_last_fetch(code) print(f"成功 ({len(processed)} 个数据点)") success += 1 # Add delay to avoid hitting rate limit if i % 10 == 0: import time time.sleep(1) # Final summary print("-" * 70) print("任务完成！") print("我辛苦了！") print("-" * 70) print(f"总股票数: {total}") print(f"成功获取: {success}") print(f"失败: {failed}") print(f"数据保存到: {DATA_DIR}") print("-" * 70) print("数据库已更新最后获取时间戳") print("=" * 70) def fetch_all_stocks_original(): """Original function to fetch all stocks (backward compatibility)""" conn = get_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT code, name, market FROM stocks WHERE status = 'active' ORDER BY code """) stocks = cursor.fetchall() conn.close() fetch_stocks(stocks) def main(): """Main function with command line arguments""" parser = argparse.ArgumentParser( description=f'XI Stock {DATA_TYPE.capitalize()} Data Fetcher - Enhanced Version', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=f""" 使用示例: 传统用法: month.py - 获取所有活跃股票数据增强用法: month.py get 000001 - 获取单只股票数据 month.py get 000001,000002 - 获取多只股票数据 month.py get all - 获取所有未更新股票 month.py get all --limit 10 - 获取10只未更新股票 month.py get rand - 随机获取5只未更新股票 month.py get rand --limit 3 - 随机获取3只未更新股票 month.py get - 获取5只未更新股票（默认） """ ) subparsers = parser.add_subparsers(dest='command', help='命令') # get command get_parser = subparsers.add_parser('get', help='获取股票数据') get_parser.add_argument('ac', nargs='?', default='', help='股票代码(all/rand/股票代码)') get_parser.add_argument('--limit', type=int, default=DEFAULT_LIMIT, help=f'获取股票数量限制 (默认: {DEFAULT_LIMIT})') args = parser.parse_args() try: if not args.command: # 传统用法：没有参数时获取所有股票 print(f"使用传统模式：获取所有活跃股票{DATA_TYPE}数据...") fetch_all_stocks_original() elif args.command == 'get': ac = args.ac.strip().lower() if args.ac else '' limit = args.limit if not ac: # 默认：获取5只未更新股票 print(f"未指定股票，获取 {limit} 只未更新股票...") stocks = get_all_stocks_to_update(limit) fetch_stocks(stocks) elif ac == 'all': # 获取所有未更新股票 if limit > 0: print(f"获取 {limit} 只未更新股票...") stocks = get_all_stocks_to_update(limit) else: print("获取所有未更新股票...") stocks = get_all_stocks_to_update() fetch_stocks(stocks) elif ac == 'rand': # 随机获取未更新股票 print(f"随机获取 {limit} 只未更新股票...") stocks = get_random_stocks_to_update(limit) fetch_stocks(stocks) else: # 获取指定股票代码 print(f"获取指定股票: {ac}") stocks = get_stocks_by_codes(ac) if not stocks: print(f"未找到股票代码: {ac}") return fetch_stocks(stocks) except KeyboardInterrupt: print("\n\n- 用户中断") exit(1) except Exception as e: print(f"\n- 错误: {e}") import traceback traceback.print_exc() exit(1) if __name__ == "__main__": main() FILE:scripts/month_original.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ XI Stock Monthly Data Fetcher 羲股票监控系统 - 月线数据获取 This script fetches monthly K-line data for all stocks from Tencent Finance API. 本脚本从腾讯财经API获取所有股票的月K线数据。 - Monthly fetching for long-term trend analysis - 月线获取用于长期趋势分析 - Saves 800 historical data points per stock - 每只股票保存800个历史数据点 """ import os import sys import sqlite3 import requests import json from datetime import datetime # Fix encoding for Windows console # 修复Windows控制台编码问题 if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" DATA_DIR = "D:\\xistock\\month" BASE_URL = "https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get" DATA_POINTS = 800 FREQ = "month" def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- Database not found: {DB_PATH}") print(" Please run init_db.py and fetch_stocks.py first!") exit(1) return sqlite3.connect(DB_PATH) def get_stocks_to_fetch(): """Get list of stocks that need to be fetched (all active stocks)""" conn = get_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT code, name, market FROM stocks WHERE status = 'active' ORDER BY code """) stocks = cursor.fetchall() conn.close() return stocks def format_stock_code(code): """Format stock code for Tencent Finance API (sh600001, sz000001)""" if code.startswith('60'): return f"sh{code}" else: return f"sz{code}" def fetch_stock_data(stock_code, stock_name): """Fetch monthly K-line data from Tencent Finance API""" try: tencent_code = format_stock_code(stock_code) params = f"{tencent_code},{FREQ},,,{DATA_POINTS},qfq" url = f"{BASE_URL}?_var=kline_monthqfq¶m={params}" response = requests.get(url, timeout=30) response.encoding = 'utf-8' if response.status_code != 200: print(f"- {stock_code} {stock_name}: HTTP {response.status_code}") return None # Response is in format: kline_monthqfq = {...}; content = response.text if '=' in content: json_str = content.split('=', 1)[1].rstrip(';') else: json_str = content data = json.loads(json_str) # Data structure: data -> tencent_code -> qfqmonth if 'data' not in data or tencent_code not in data['data']: print(f"- {stock_code} {stock_name}: No data found") return None stock_data = data['data'][tencent_code] if 'qfqmonth' not in stock_data: print(f"- {stock_code} {stock_name}: No qfqmonth found") return None return stock_data['qfqmonth'] except Exception as e: print(f"- {stock_code} {stock_name}: Error fetching - {str(e)}") return None def process_data(data): """ Process raw data, extract date, open, close, high, low and calculate change 返回格式列表: [date, open, close, high, low, change] """ processed = [] # Data format from Tencent: [date, open, close, high, low, volume, ...] for bar in data: if len(bar) >= 5: date = bar[0] open_p = float(bar[1]) close = float(bar[2]) high = float(bar[3]) low = float(bar[4]) processed.append([date, open_p, close, high, low]) # Calculate change (涨跌幅) for i in range(len(processed)): if i == 0: change = 0.0 else: prev_close = processed[i-1][2] if prev_close == 0: change = 0.0 else: change = ((processed[i][2] - prev_close) / prev_close) * 100 processed[i].append(round(change, 3)) return processed def save_to_file(stock_code, stock_name, data): """ Save processed data to file, one data point per line Format: date,open,close,high,low,change Filename: 名称_代码.txt """ # Create directory if not exists if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) # Sanitize filename - remove all invalid Windows filename characters # Invalid chars: \ / : * ? " < > | invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = stock_name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{stock_code}.txt" filepath = os.path.join(DATA_DIR, filename) # Write header with open(filepath, 'w', encoding='utf-8') as f: f.write("date,open,close,high,low,change_pct\n") for item in data: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) return filepath def update_database_last_fetch(stock_code): """Update the month_get timestamp in database to current date""" conn = get_db_connection() cursor = conn.cursor() current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') cursor.execute(""" UPDATE stocks SET month_get = ? WHERE code = ? """, (current_time, stock_code)) conn.commit() conn.close() def fetch_all_stocks(): """Fetch data for all stocks""" stocks = get_stocks_to_fetch() total = len(stocks) print("=" * 70) print("XI Stock Monthly Data Fetcher") print("羲股票监控系统 - 月线数据获取") print("=" * 70) print(f"Total active stocks: {total}") print(f"Data directory: {DATA_DIR}") print("-" * 70) success = 0 failed = 0 for i, (code, name, market) in enumerate(stocks, 1): print(f"[{i}/{total}] Fetching {code} {name}...", end=' ') sys.stdout.flush() # Fetch data raw_data = fetch_stock_data(code, name) if raw_data is None: print("FAILED") failed += 1 continue # Process data bars = raw_data processed = process_data(bars) if not processed: print("NO DATA") failed += 1 continue # Save to file save_to_file(code, name, processed) # Update database update_database_last_fetch(code) print(f"OK ({len(processed)} points)") success += 1 # Add delay to avoid hitting rate limit if i % 10 == 0: import time time.sleep(1) # Final summary print("-" * 70) print("任务完成！") print("我辛苦了！") print("-" * 70) print(f"Total stocks: {total}") print(f"Successfully fetched: {success}") print(f"Failed: {failed}") print(f"Data saved to: {DATA_DIR}") print("-" * 70) print("Database updated with last fetch timestamps") print("=" * 70) if __name__ == "__main__": try: fetch_all_stocks() except KeyboardInterrupt: print("\n\n- Interrupted by user") exit(1) except Exception as e: print(f"\n- Error: {e}") import traceback traceback.print_exc() exit(1) FILE:scripts/month_parallel.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ A-Share-Get Monthly Data Fetcher - Parallel Incremental Version A股数据获取 - 月线数据增量获取（并行版本） Usage: python scripts/month_parallel.py asc # 获取前一半（升序），每次2-3只 python scripts/month_parallel.py desc # 获取后一半（降序），每次2-3只 Run both in separate processes to speed up incremental updates! 在两个独立进程中同时运行，加速增量更新！ Incremental update logic: - Read existing file, get latest date - Only fetch data newer than latest date (add to file, don't overwrite) - If file doesn't exist, fetch full 800 points - Process max 100 stocks per run (incremental) / 800 (new file) - Update database timestamp after successful fetch """ import os import sys import sqlite3 import requests import json from datetime import datetime # Fix encoding for Windows console if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" DATA_DIR = "D:\\xistock\\month" BASE_URL = "https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get" DATA_POINTS_FULL = 800 # When file doesn't exist MAX_STOCKS_PER_RUN_INCREMENTAL = 100 # Max stocks to process per run when incremental update FREQ = "month" BATCH_SIZE = 3 # Process 2-3 stocks per batch to avoid long runtime def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- Database not found: {DB_PATH}") print(" Please run init_db.py and fetch_stocks.py first!") exit(1) return sqlite3.connect(DB_PATH) def get_latest_date_from_file(filepath): """ Read existing file and get the latest date Returns None if file doesn't exist or empty """ if not os.path.exists(filepath): return None try: with open(filepath, 'r', encoding='utf-8') as f: # Skip header next(f) latest_date = None # Read all lines to get the last date for line in f: line = line.strip() if not line: continue date = line.split(',')[0] latest_date = date return latest_date except Exception as e: print(f" Warning: Cannot read existing file: {e}") return None def get_stocks_to_fetch(order): """ Get list of stocks that need to be fetched (incremental) - Selects stocks that have never been fetched or haven't been fetched recently - order: 'asc' or 'desc' - split into two partitions - Returns up to MAX_STOCKS_PER_RUN_INCREMENTAL stocks """ conn = get_db_connection() cursor = conn.cursor() # Get all active stocks ordered by code cursor.execute(""" SELECT code, name, market, month_get FROM stocks WHERE status = 'active' ORDER BY code """) all_stocks = cursor.fetchall() total = len(all_stocks) half = total // 2 # Split into two partitions if order == 'asc': partition_stocks = all_stocks[:half] elif order == 'desc': all_stocks.reverse() partition_stocks = all_stocks[:half] else: print(f"- Invalid order: {order}. Use 'asc' or 'desc'") exit(1) # Filter stocks that need update (any stock that hasn't been fetched today) today = datetime.now().strftime('%Y-%m-%d') need_update = [] for stock in partition_stocks: if len(stock) == 3: code, name, market = stock month_get = None else: code, name, market, month_get = stock # If never fetched or last fetched not today, needs update if month_get is None or not month_get.startswith(today): need_update.append((code, name, market)) # Process all need update today, large safety limit to prevent infinite run # MAX_STOCKS_PER_RUN_INCREMENTAL is original limit, now we allow 10x that if len(need_update) > MAX_STOCKS_PER_RUN_INCREMENTAL * 10: need_update = need_update[:MAX_STOCKS_PER_RUN_INCREMENTAL * 10] print(f"Partition {order}:") print(f" Total in partition: {len(partition_stocks)}") print(f" Need update today: {len(need_update)}") print(f" Will process all {len(need_update)} stocks (safety limit: {MAX_STOCKS_PER_RUN_INCREMENTAL * 10})") print(f" Will process: {len(need_update)} (max {MAX_STOCKS_PER_RUN_INCREMENTAL})") conn.close() return need_update def format_stock_code(code): """Format stock code for Tencent Finance API (sh600001, sz000001)""" if code.startswith('60'): return f"sh{code}" else: return f"sz{code}" def fetch_stock_data(stock_code, stock_name): """Fetch monthly K-line data from Tencent Finance API""" try: tencent_code = format_stock_code(stock_code) params = f"{tencent_code},{FREQ},,,{DATA_POINTS_FULL},qfq" url = f"{BASE_URL}?_var=kline_monthqfq¶m={params}" response = requests.get(url, timeout=30) response.encoding = 'utf-8' if response.status_code != 200: print(f"- {stock_code} {stock_name}: HTTP {response.status_code}") return None # Response is in format: kline_monthqfq = {...}; content = response.text if '=' in content: json_str = content.split('=', 1)[1].rstrip(';') else: json_str = content data = json.loads(json_str) # Data structure: data -> tencent_code -> qfqmonth if 'data' not in data or tencent_code not in data['data']: print(f"- {stock_code} {stock_name}: No data found") return None stock_data = data['data'][tencent_code] if 'qfqmonth' not in stock_data: print(f"- {stock_code} {stock_name}: No qfqmonth found") return None return stock_data['qfqmonth'] except Exception as e: print(f"- {stock_code} {stock_name}: Error fetching - {str(e)}") return None def process_data(data): """ Process raw data, extract date, open, close, high, low and calculate change 返回格式列表: [date, open, close, high, low, change] """ processed = [] # Data format from Tencent: [date, open, close, high, low, volume, ...] for bar in data: if len(bar) >= 5: date = bar[0] open_p = float(bar[1]) close = float(bar[2]) high = float(bar[3]) low = float(bar[4]) processed.append([date, open_p, close, high, low]) # Calculate change (涨跌幅) for i in range(len(processed)): if i == 0: change = 0.0 else: prev_close = processed[i-1][2] if prev_close == 0: change = 0.0 else: change = ((processed[i][2] - prev_close) / prev_close) * 100 processed[i].append(round(change, 3)) return processed def get_data_points_to_fetch(processed_new, latest_date): """ Filter new data points that are after latest_date in existing file If latest_date is None, return all """ if latest_date is None: return processed_new, DATA_POINTS_FULL # Find all data points after latest_date new_data = [] for item in processed_new: date = item[0] if date > latest_date: new_data.append(item) return new_data, len(processed_new) def append_to_file(stock_code, stock_name, new_data, latest_date): """ Append new data points to existing file, don't overwrite If file doesn't exist, create new with header Format: date,open,close,high,low,change Filename: 名称_代码.txt """ # Create directory if not exists if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) # Sanitize filename - remove all invalid Windows filename characters invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = stock_name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{stock_code}.txt" filepath = os.path.join(DATA_DIR, filename) # Filter new data (only data after latest_date) data_to_write, total_fetched = get_data_points_to_fetch(new_data, latest_date) if not data_to_write: # No new data to add return filepath, 0 if latest_date is None: # Create new file with open(filepath, 'w', encoding='utf-8') as f: f.write("date,open,close,high,low,change_pct\n") for item in data_to_write: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) else: # Append to existing file with open(filepath, 'a', encoding='utf-8') as f: for item in data_to_write: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) return filepath, len(data_to_write) def update_database_last_fetch(stock_code): """Update the month_get timestamp in database to current date""" conn = get_db_connection() cursor = conn.cursor() current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') cursor.execute(""" UPDATE stocks SET month_get = ? WHERE code = ? """, (current_time, stock_code)) conn.commit() conn.close() def fetch_all_stocks(order): """Fetch data incrementally for stocks in this partition - process {BATCH_SIZE} per run""" stocks = get_stocks_to_fetch(order) total_need_update = len(stocks) print("=" * 70) print(f"A-Share-Get Monthly Data Fetcher - Incremental Parallel ({order})") print("A股数据获取 - 月线增量更新（并行）") print("=" * 70) print(f"Data directory: {DATA_DIR}") print(f"Processing {BATCH_SIZE} stocks per run (max {MAX_STOCKS_PER_RUN_INCREMENTAL} total)") print("-" * 70) success_total = 0 failed_total = 0 new_points_total = 0 # Process in batches of BATCH_SIZE from itertools import islice def batch_generator(iterable, batch_size): iterator = iter(iterable) for first in iterator: yield [first] + list(islice(iterator, batch_size - 1)) for batch_idx, batch in enumerate(batch_generator(stocks, BATCH_SIZE), 1): print(f"\n>> Batch {batch_idx}, processing {len(batch)} stocks:") for i, (code, name, market) in enumerate(batch, 1): global_i = (batch_idx - 1) * BATCH_SIZE + i print(f" [{global_i}/{total_need_update}] {code} {name}: ", end=' ') sys.stdout.flush() # Get existing file's latest date invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{code}.txt" filepath = os.path.join(DATA_DIR, filename) latest_date = get_latest_date_from_file(filepath) if latest_date: print(f"latest={latest_date}, fetching new data...", end=' ') else: print("new file, fetching full data...", end=' ') sys.stdout.flush() # Fetch data raw_data = fetch_stock_data(code, name) if raw_data is None: print("FAILED") failed_total += 1 continue # Process data processed = process_data(raw_data) if not processed: print("NO DATA") failed_total += 1 continue # Append new data to file (only add missing) filepath, new_count = append_to_file(code, name, processed, latest_date) if new_count == 0: print("no new data") # Still update database to mark as updated today update_database_last_fetch(code) continue # Update database update_database_last_fetch(code) print(f"OK added {new_count} new points") success_total += 1 new_points_total += new_count # Add delay to avoid hitting rate limit import time time.sleep(1.5) # Final summary print("-" * 70) print("增量更新完成！") print("-" * 70) print(f"Partition ({order}):") print(f" - Need update: {total_need_update}") print(f" - Successfully processed: {success_total}") print(f" - Failed: {failed_total}") print(f" - New data points added: {new_points_total}") print(f"Data directory: {DATA_DIR}") print("-" * 70) print("Database updated with last fetch timestamps") print("=" * 70) if __name__ == "__main__": if len(sys.argv) != 2: print("Usage:") print(" python scripts/month_parallel.py asc # First half (ascending)") print(" python scripts/month_parallel.py desc # Second half (descending)") print("\nRun both in separate processes for parallel incremental fetching!") exit(1) order = sys.argv[1].lower() try: fetch_all_stocks(order) except KeyboardInterrupt: print("\n\n- Interrupted by user") exit(1) except Exception as e: print(f"\n- Error: {e}") import traceback traceback.print_exc() exit(1) FILE:scripts/README.md # XI Stock Data Fetcher - Enhanced System # 羲股票监控系统 - 增强版数据获取 ## 📋 核心文件 ### 主脚本 - `day.py` - 增强版日线数据获取（支持外部事件） - `week.py` - 增强版周线数据获取（支持外部事件） - `month.py` - 增强版月线数据获取（支持外部事件） - `db_reset.py` - 数据库重置与数据获取工具 ### 并行版本（批量处理） - `day_parallel.py` - 并行日线数据获取 - `week_parallel.py` - 并行周线数据获取 - `month_parallel.py` - 并行月线数据获取 ### 基础脚本 - `init_db.py` - 数据库初始化 - `fetch_stocks.py` - 获取股票列表 ### 备份文件 - `day_original.py` - 原始日线获取脚本备份 - `week_original.py` - 原始周线获取脚本备份 - `month_original.py` - 原始月线获取脚本备份 ## 🚀 快速使用 ### 传统用法（获取所有活跃股票） ```bash python day.py python week.py python month.py ``` ### 增强用法（支持外部事件） #### 获取单只股票 ```bash python day.py get 000001 python week.py get 000001 python month.py get 000001 ``` #### 获取多只股票（逗号分隔） ```bash python day.py get 000001,000002 python week.py get 000001,000002 python month.py get 000001,000002 ``` #### 获取所有未更新股票 ```bash python day.py get all python week.py get all python month.py get all # 限制数量 python day.py get all --limit 10 ``` #### 随机获取股票 ```bash python day.py get rand python week.py get rand python month.py get rand # 指定数量 python day.py get rand --limit 3 ``` #### 默认行为（获取5只未更新股票） ```bash python day.py get python week.py get python month.py get ``` ### 数据库管理工具 #### 重置功能 ```bash python db_reset.py reset day # 重置日线时间戳 python db_reset.py reset week # 重置周线时间戳 python db_reset.py reset month # 重置月线时间戳 python db_reset.py reset all # 重置所有时间戳 python db_reset.py reset status # 查看数据库状态 ``` #### 直接获取功能 ```bash # 日线数据获取 python db_reset.py fetch day 000001 python db_reset.py fetch day all --limit 5 python db_reset.py fetch day rand --limit 3 # 周线数据获取 python db_reset.py fetch week 000001 python db_reset.py fetch week all --limit 5 # 月线数据获取 python db_reset.py fetch month 000001 python db_reset.py fetch month all --limit 5 ``` ## 📊 智能更新逻辑 1. **增量更新**：只获取 `timestamp_field < 当前时间` 的股票 2. **优先级**：从未更新的股票（NULL）优先 3. **随机选择**：`rand` 模式用于分散更新压力 4. **批量处理**：支持逗号分隔的多股票代码 5. **数量限制**：`--limit` 参数控制获取数量 ## 🗄️ 数据库字段 ### stocks 表结构 | 字段 | 类型 | 说明 | |------|------|------| | code | TEXT | 股票代码（主键） | | name | TEXT | 股票名称 | | market | TEXT | 市场类型 | | day_get | TIMESTAMP | 最后日线获取时间 | | week_get | TIMESTAMP | 最后周线获取时间 | | month_get | TIMESTAMP | 最后月线获取时间 | | status | TEXT | 状态（active/inactive） | | created_at | TIMESTAMP | 创建时间 | ## 📁 数据文件位置 - **日线数据**: `D:\xistock\day\股票名称_股票代码.txt` - **周线数据**: `D:\xistock\week\股票名称_股票代码.txt` - **月线数据**: `D:\xistock\month\股票名称_股票代码.txt` ### 文件格式 ``` date,open,close,high,low,change_pct 2024-01-01,10.50,10.80,11.00,10.40,2.857 2024-01-02,10.85,10.70,11.10,10.60,-0.926 ... ``` ## ⚙️ 初始化步骤 1. **初始化数据库** ```bash python init_db.py ``` 2. **获取股票列表** ```bash python fetch_stocks.py ``` 3. **开始数据收集** ```bash # 传统方式（获取所有股票） python day.py python week.py python month.py # 或使用增强方式 python day.py get all --limit 10 ``` ## ⚠️ 注意事项 1. **API限制**：腾讯财经API有速率限制，脚本包含延迟机制 2. **网络连接**：需要稳定的网络连接 3. **磁盘空间**：确保 `D:\xistock` 有足够空间 4. **交易时段**：建议在非交易时段运行 5. **数据完整性**：重置操作会标记所有股票为需要更新 ## 🔧 故障排除 ### 数据库不存在 ```bash # 运行初始化脚本 python init_db.py python fetch_stocks.py ``` ### 股票代码不存在 ```bash # 检查股票是否在数据库中 python fetch_stocks.py # 更新股票列表 ``` ### 网络连接失败 - 检查网络连接 - 脚本会自动重试失败股票 - 可以单独重试失败股票 ## 📈 性能建议 1. **批量处理**：使用逗号分隔一次获取多只股票 2. **合理限制**：根据网络情况设置合适的 `--limit` 值 3. **定时任务**：配置Windows任务计划或Linux cron 4. **并行处理**：对于大量股票，使用 `*_parallel.py` 脚本 --- **版本**: 2.0.0 (增强版) **最后更新**: 2026-03-14 **功能**: 支持外部事件控制、智能更新、批量处理 FILE:scripts/schedule_config.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Schedule Configuration for XI Stock System 羲股票监控系统定时任务配置 This script provides configuration for OpenClaw cron jobs to schedule automatic updates. 本脚本提供OpenClaw定时任务的配置，用于安排自动更新。 """ import json from datetime import datetime, timedelta def generate_cron_config(): """Generate cron job configuration for OpenClaw""" config = { "name": "XI Stock System - Daily Updates", "description": "Automated stock data fetching for A-share market", "jobs": [ { "name": "Daily Stock Data Update", "description": "Fetch daily data for 10 stocks each time", "schedule": { "kind": "cron", "expr": "30 15 * * 1-5", # 工作日15:30（收盘后） "tz": "Asia/Shanghai" }, "payload": { "kind": "agentTurn", "message": "Run daily stock data update: python day.py get --limit 10", "model": "deepseek-v3.2" }, "sessionTarget": "isolated", "delivery": { "mode": "announce", "channel": "telegram" } }, { "name": "Weekly Reset and Update", "description": "Reset week_get and fetch weekly data on Monday", "schedule": { "kind": "cron", "expr": "0 9 * * 1", # 周一9:00 "tz": "Asia/Shanghai" }, "payload": { "kind": "agentTurn", "message": "Reset week_get and fetch weekly data: python db_reset.py reset week && python week.py get all --limit 20", "model": "deepseek-v3.2" }, "sessionTarget": "isolated", "delivery": { "mode": "announce", "channel": "telegram" } }, { "name": "Monthly Reset and Update", "description": "Reset month_get and fetch monthly data on 1st of month", "schedule": { "kind": "cron", "expr": "0 9 1 * *", # 每月1日9:00 "tz": "Asia/Shanghai" }, "payload": { "kind": "agentTurn", "message": "Reset month_get and fetch monthly data: python db_reset.py reset month && python month.py get all --limit 20", "model": "deepseek-v3.2" }, "sessionTarget": "isolated", "delivery": { "mode": "announce", "channel": "telegram" } }, { "name": "Database Status Check", "description": "Check database status every day at 10:00", "schedule": { "kind": "cron", "expr": "0 10 * * *", # 每天10:00 "tz": "Asia/Shanghai" }, "payload": { "kind": "agentTurn", "message": "Check database status: python db_reset.py reset status", "model": "deepseek-v3.2" }, "sessionTarget": "isolated", "delivery": { "mode": "announce", "channel": "telegram" } } ] } return config def generate_heartbeat_config(): """Generate heartbeat configuration for periodic checks""" config = { "name": "XI Stock System - Heartbeat Checks", "description": "Periodic checks during heartbeats", "checks": [ { "name": "Check Database Status", "command": "python db_reset.py reset status", "frequency": "every_heartbeat", # 每次心跳时检查 "condition": "always" }, { "name": "Check if Need Daily Update", "command": "python day.py get --limit 1 --test-only", "frequency": "every_4_hours", "condition": "if_business_hours" }, { "name": "Check Data Directory", "command": "dir D:\\xistock\\day /b | find /c \".txt\"", "frequency": "daily", "condition": "always" } ] } return config def generate_quick_commands(): """Generate quick commands for manual execution""" commands = { "daily_update_10": "python day.py get --limit 10", "daily_update_all": "python day.py get all", "weekly_update": "python week.py get all --limit 20", "monthly_update": "python month.py get all --limit 20", "reset_all": "python db_reset.py reset all", "status": "python db_reset.py reset status", "fetch_single": "python db_reset.py fetch day 000001", "fetch_random": "python db_reset.py fetch day rand --limit 5", "parallel_day": "python day_parallel.py", "parallel_week": "python week_parallel.py", "parallel_month": "python month_parallel.py" } return commands def main(): """Main function to display configuration""" print("=" * 70) print("XI Stock System - Schedule Configuration") print("羲股票监控系统 - 定时任务配置") print("=" * 70) # Generate configurations cron_config = generate_cron_config() heartbeat_config = generate_heartbeat_config() quick_commands = generate_quick_commands() print("\n📅 Cron Job Configuration (for OpenClaw):") print("-" * 70) for job in cron_config["jobs"]: print(f" • {job['name']}") print(f" 时间: {job['schedule']['expr']} ({job['schedule']['tz']})") print(f" 命令: {job['payload']['message']}") print() print("\n💓 Heartbeat Checks:") print("-" * 70) for check in heartbeat_config["checks"]: print(f" • {check['name']}") print(f" 频率: {check['frequency']}") print(f" 条件: {check['condition']}") print(f" 命令: {check['command']}") print() print("\n⚡ Quick Commands (for manual execution):") print("-" * 70) for name, cmd in quick_commands.items(): print(f" {name:20} → {cmd}") print("\n📋 To add cron jobs to OpenClaw:") print("-" * 70) print("1. Save this configuration to a file") print("2. Use OpenClaw cron tool to add jobs:") print(" openclaw cron add --config schedule_config.json") print("\n3. Or add manually using OpenClaw web interface") print("=" * 70) # Save configuration to files with open("schedule_config.json", "w", encoding="utf-8") as f: json.dump(cron_config, f, ensure_ascii=False, indent=2) with open("quick_commands.json", "w", encoding="utf-8") as f: json.dump(quick_commands, f, ensure_ascii=False, indent=2) print("\n✅ Configuration files saved:") print(" - schedule_config.json") print(" - quick_commands.json") if __name__ == "__main__": main() FILE:scripts/week.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ XI Stock Weekly Data Fetcher - Enhanced Version 羲股票监控系统 - 周线数据获取（增强版） This script fetches weekly K-line data for stocks from Tencent Finance API. 本脚本从腾讯财经API获取股票的周K线数据。新增功能： week.py get [ac] - 获取股票数据 - ac: 股票代码（支持逗号分隔多个股票） - all: 获取所有未更新股票 - rand: 随机获取5只未更新股票 - 默认：获取5只未更新股票 """ import os import sys import sqlite3 import requests import json import random import argparse from datetime import datetime, timedelta # Fix encoding for Windows console # 修复Windows控制台编码问题 if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" DATA_DIR = "D:\\xistock\\week" BASE_URL = "https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get" DATA_POINTS = 800 FREQ = "week" DEFAULT_LIMIT = 5 TIMESTAMP_FIELD = "week_get" DATA_TYPE = "周线" def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- 数据库未找到: {DB_PATH}") print(" 请先运行 init_db.py 和 fetch_stocks.py!") exit(1) return sqlite3.connect(DB_PATH) def get_stocks_by_codes(stock_codes): """Get stocks by specific codes""" conn = get_db_connection() cursor = conn.cursor() # 处理逗号分隔的股票代码 codes = [code.strip() for code in stock_codes.split(',')] # 构建查询 placeholders = ','.join(['?'] * len(codes)) query = f""" SELECT code, name, market FROM stocks WHERE code IN ({placeholders}) AND status = 'active' ORDER BY code """ cursor.execute(query, codes) stocks = cursor.fetchall() conn.close() return stocks def get_all_stocks_to_update(limit=None): """Get all stocks that need to be updated (week_get is NULL or old)""" conn = get_db_connection() cursor = conn.cursor() # 获取当前时间 current_time = datetime.now() query = f""" SELECT code, name, market FROM stocks WHERE status = 'active' AND ({TIMESTAMP_FIELD} IS NULL OR {TIMESTAMP_FIELD} < ?) ORDER BY {TIMESTAMP_FIELD} ASC NULLS FIRST """ cursor.execute(query, (current_time,)) stocks = cursor.fetchall() conn.close() # 应用限制 if limit and limit > 0: stocks = stocks[:limit] return stocks def get_random_stocks_to_update(limit=5): """Get random stocks that need to be updated""" conn = get_db_connection() cursor = conn.cursor() # 获取当前时间 current_time = datetime.now() query = f""" SELECT code, name, market FROM stocks WHERE status = 'active' AND ({TIMESTAMP_FIELD} IS NULL OR {TIMESTAMP_FIELD} < ?) """ cursor.execute(query, (current_time,)) all_stocks = cursor.fetchall() conn.close() # 随机选择 if len(all_stocks) > limit: stocks = random.sample(all_stocks, limit) else: stocks = all_stocks return stocks def format_stock_code(code): """Format stock code for Tencent Finance API (sh600001, sz000001)""" if code.startswith('60'): return f"sh{code}" else: return f"sz{code}" def fetch_stock_data(stock_code, stock_name): """Fetch weekly K-line data from Tencent Finance API""" try: tencent_code = format_stock_code(stock_code) params = f"{tencent_code},{FREQ},,,{DATA_POINTS},qfq" url = f"{BASE_URL}?_var=kline_{FREQ}qfq¶m={params}" response = requests.get(url, timeout=30) response.encoding = 'utf-8' if response.status_code != 200: print(f"- {stock_code} {stock_name}: HTTP {response.status_code}") return None # Response is in format: kline_weekqfq = {...}; content = response.text if '=' in content: json_str = content.split('=', 1)[1].rstrip(';') else: json_str = content data = json.loads(json_str) # Data structure: data -> tencent_code -> qfqweek if 'data' not in data or tencent_code not in data['data']: print(f"- {stock_code} {stock_name}: No data found") return None stock_data = data['data'][tencent_code] data_key = f"qfq{FREQ}" if data_key not in stock_data: print(f"- {stock_code} {stock_name}: No {data_key} found") return None return stock_data[data_key] except Exception as e: print(f"- {stock_code} {stock_name}: Error fetching - {str(e)}") return None def process_data(data): """ Process raw data, extract date, open, close, high, low and calculate change 返回格式列表: [date, open, close, high, low, change] """ processed = [] # Data format from Tencent: [date, open, close, high, low, volume, ...] for bar in data: if len(bar) >= 5: date = bar[0] open_p = float(bar[1]) close = float(bar[2]) high = float(bar[3]) low = float(bar[4]) processed.append([date, open_p, close, high, low]) # Calculate change (涨跌幅) for i in range(len(processed)): if i == 0: change = 0.0 else: prev_close = processed[i-1][2] if prev_close == 0: change = 0.0 else: change = ((processed[i][2] - prev_close) / prev_close) * 100 processed[i].append(round(change, 3)) return processed def save_to_file(stock_code, stock_name, data): """ Save processed data to file, one data point per line Format: date,open,close,high,low,change Filename: 名称_代码.txt """ # Create directory if not exists if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) # Sanitize filename - remove all invalid Windows filename characters # Invalid chars: \ / : * ? " < > | invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = stock_name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{stock_code}.txt" filepath = os.path.join(DATA_DIR, filename) # Write header with open(filepath, 'w', encoding='utf-8') as f: f.write("date,open,close,high,low,change_pct\n") for item in data: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) return filepath def update_database_last_fetch(stock_code): """Update the week_get timestamp in database to current date""" conn = get_db_connection() cursor = conn.cursor() current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') cursor.execute(f""" UPDATE stocks SET {TIMESTAMP_FIELD} = ? WHERE code = ? """, (current_time, stock_code)) conn.commit() conn.close() def fetch_stocks(stocks): """Fetch data for given list of stocks""" total = len(stocks) if total == 0: print("没有需要更新的股票") return print("=" * 70) print(f"XI Stock {DATA_TYPE.capitalize()} Data Fetcher") print(f"羲股票监控系统 - {DATA_TYPE}数据获取") print("=" * 70) print(f"股票数量: {total}") print(f"数据目录: {DATA_DIR}") print("-" * 70) success = 0 failed = 0 for i, (code, name, market) in enumerate(stocks, 1): print(f"[{i}/{total}] 获取 {code} {name}...", end=' ') sys.stdout.flush() # Fetch data raw_data = fetch_stock_data(code, name) if raw_data is None: print("失败") failed += 1 continue # Process data bars = raw_data processed = process_data(bars) if not processed: print("无数据") failed += 1 continue # Save to file save_to_file(code, name, processed) # Update database update_database_last_fetch(code) print(f"成功 ({len(processed)} 个数据点)") success += 1 # Add delay to avoid hitting rate limit if i % 10 == 0: import time time.sleep(1) # Final summary print("-" * 70) print("任务完成！") print("我辛苦了！") print("-" * 70) print(f"总股票数: {total}") print(f"成功获取: {success}") print(f"失败: {failed}") print(f"数据保存到: {DATA_DIR}") print("-" * 70) print("数据库已更新最后获取时间戳") print("=" * 70) def fetch_all_stocks_original(): """Original function to fetch all stocks (backward compatibility)""" conn = get_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT code, name, market FROM stocks WHERE status = 'active' ORDER BY code """) stocks = cursor.fetchall() conn.close() fetch_stocks(stocks) def main(): """Main function with command line arguments""" parser = argparse.ArgumentParser( description=f'XI Stock {DATA_TYPE.capitalize()} Data Fetcher - Enhanced Version', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=f""" 使用示例: 传统用法: week.py - 获取所有活跃股票数据增强用法: week.py get 000001 - 获取单只股票数据 week.py get 000001,000002 - 获取多只股票数据 week.py get all - 获取所有未更新股票 week.py get all --limit 10 - 获取10只未更新股票 week.py get rand - 随机获取5只未更新股票 week.py get rand --limit 3 - 随机获取3只未更新股票 week.py get - 获取5只未更新股票（默认） """ ) subparsers = parser.add_subparsers(dest='command', help='命令') # get command get_parser = subparsers.add_parser('get', help='获取股票数据') get_parser.add_argument('ac', nargs='?', default='', help='股票代码(all/rand/股票代码)') get_parser.add_argument('--limit', type=int, default=DEFAULT_LIMIT, help=f'获取股票数量限制 (默认: {DEFAULT_LIMIT})') args = parser.parse_args() try: if not args.command: # 传统用法：没有参数时获取所有股票 print(f"使用传统模式：获取所有活跃股票{DATA_TYPE}数据...") fetch_all_stocks_original() elif args.command == 'get': ac = args.ac.strip().lower() if args.ac else '' limit = args.limit if not ac: # 默认：获取5只未更新股票 print(f"未指定股票，获取 {limit} 只未更新股票...") stocks = get_all_stocks_to_update(limit) fetch_stocks(stocks) elif ac == 'all': # 获取所有未更新股票 if limit > 0: print(f"获取 {limit} 只未更新股票...") stocks = get_all_stocks_to_update(limit) else: print("获取所有未更新股票...") stocks = get_all_stocks_to_update() fetch_stocks(stocks) elif ac == 'rand': # 随机获取未更新股票 print(f"随机获取 {limit} 只未更新股票...") stocks = get_random_stocks_to_update(limit) fetch_stocks(stocks) else: # 获取指定股票代码 print(f"获取指定股票: {ac}") stocks = get_stocks_by_codes(ac) if not stocks: print(f"未找到股票代码: {ac}") return fetch_stocks(stocks) except KeyboardInterrupt: print("\n\n- 用户中断") exit(1) except Exception as e: print(f"\n- 错误: {e}") import traceback traceback.print_exc() exit(1) if __name__ == "__main__": main() FILE:scripts/week_original.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ XI Stock Weekly Data Fetcher 羲股票监控系统 - 周线数据获取 This script fetches weekly K-line data for all stocks from Tencent Finance API. 本脚本从腾讯财经API获取所有股票的周K线数据。 - Weekly fetching for long-term trend analysis - 周线获取用于长期趋势分析 - Saves 800 historical data points per stock - 每只股票保存800个历史数据点 """ import os import sys import sqlite3 import requests import json from datetime import datetime # Fix encoding for Windows console # 修复Windows控制台编码问题 if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" DATA_DIR = "D:\\xistock\\week" BASE_URL = "https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get" DATA_POINTS = 800 FREQ = "week" def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- Database not found: {DB_PATH}") print(" Please run init_db.py and fetch_stocks.py first!") exit(1) return sqlite3.connect(DB_PATH) def get_stocks_to_fetch(): """Get list of stocks that need to be fetched (all active stocks)""" conn = get_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT code, name, market FROM stocks WHERE status = 'active' ORDER BY code """) stocks = cursor.fetchall() conn.close() return stocks def format_stock_code(code): """Format stock code for Tencent Finance API (sh600001, sz000001)""" if code.startswith('60'): return f"sh{code}" else: return f"sz{code}" def fetch_stock_data(stock_code, stock_name): """Fetch weekly K-line data from Tencent Finance API""" try: tencent_code = format_stock_code(stock_code) params = f"{tencent_code},{FREQ},,,{DATA_POINTS},qfq" url = f"{BASE_URL}?_var=kline_weekqfq¶m={params}" response = requests.get(url, timeout=30) response.encoding = 'utf-8' if response.status_code != 200: print(f"- {stock_code} {stock_name}: HTTP {response.status_code}") return None # Response is in format: kline_weekqfq = {...}; content = response.text if '=' in content: json_str = content.split('=', 1)[1].rstrip(';') else: json_str = content data = json.loads(json_str) # Data structure: data -> tencent_code -> qfqweek if 'data' not in data or tencent_code not in data['data']: print(f"- {stock_code} {stock_name}: No data found") return None stock_data = data['data'][tencent_code] if 'qfqweek' not in stock_data: print(f"- {stock_code} {stock_name}: No qfqweek found") return None return stock_data['qfqweek'] except Exception as e: print(f"- {stock_code} {stock_name}: Error fetching - {str(e)}") return None def process_data(data): """ Process raw data, extract date, open, close, high, low and calculate change 返回格式列表: [date, open, close, high, low, change] """ processed = [] # Data format from Tencent: [date, open, close, high, low, volume, ...] for bar in data: if len(bar) >= 5: date = bar[0] open_p = float(bar[1]) close = float(bar[2]) high = float(bar[3]) low = float(bar[4]) processed.append([date, open_p, close, high, low]) # Calculate change (涨跌幅) for i in range(len(processed)): if i == 0: change = 0.0 else: prev_close = processed[i-1][2] if prev_close == 0: change = 0.0 else: change = ((processed[i][2] - prev_close) / prev_close) * 100 processed[i].append(round(change, 3)) return processed def save_to_file(stock_code, stock_name, data): """ Save processed data to file, one data point per line Format: date,open,close,high,low,change Filename: 名称_代码.txt """ # Create directory if not exists if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) # Sanitize filename - remove all invalid Windows filename characters # Invalid chars: \ / : * ? " < > | invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = stock_name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{stock_code}.txt" filepath = os.path.join(DATA_DIR, filename) # Write header with open(filepath, 'w', encoding='utf-8') as f: f.write("date,open,close,high,low,change_pct\n") for item in data: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) return filepath def update_database_last_fetch(stock_code): """Update the week_get timestamp in database to current date""" conn = get_db_connection() cursor = conn.cursor() current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') cursor.execute(""" UPDATE stocks SET week_get = ? WHERE code = ? """, (current_time, stock_code)) conn.commit() conn.close() def fetch_all_stocks(): """Fetch data for all stocks""" stocks = get_stocks_to_fetch() total = len(stocks) print("=" * 70) print("XI Stock Weekly Data Fetcher") print("羲股票监控系统 - 周线数据获取") print("=" * 70) print(f"Total active stocks: {total}") print(f"Data directory: {DATA_DIR}") print("-" * 70) success = 0 failed = 0 for i, (code, name, market) in enumerate(stocks, 1): print(f"[{i}/{total}] Fetching {code} {name}...", end=' ') sys.stdout.flush() # Fetch data raw_data = fetch_stock_data(code, name) if raw_data is None: print("FAILED") failed += 1 continue # Process data bars = raw_data processed = process_data(bars) if not processed: print("NO DATA") failed += 1 continue # Save to file save_to_file(code, name, processed) # Update database update_database_last_fetch(code) print(f"OK ({len(processed)} points)") success += 1 # Add delay to avoid hitting rate limit if i % 10 == 0: import time time.sleep(1) # Final summary print("-" * 70) print("任务完成！") print("我辛苦了！") print("-" * 70) print(f"Total stocks: {total}") print(f"Successfully fetched: {success}") print(f"Failed: {failed}") print(f"Data saved to: {DATA_DIR}") print("-" * 70) print("Database updated with last fetch timestamps") print("=" * 70) if __name__ == "__main__": try: fetch_all_stocks() except KeyboardInterrupt: print("\n\n- Interrupted by user") exit(1) except Exception as e: print(f"\n- Error: {e}") import traceback traceback.print_exc() exit(1) FILE:scripts/week_parallel.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ A-Share-Get Weekly Data Fetcher - Parallel Incremental Version A股数据获取 - 周线数据增量获取（并行版本） Usage: python scripts/week_parallel.py asc # 获取前一半（升序），每次2-3只 python scripts/week_parallel.py desc # 获取后一半（降序），每次2-3只 Run both in separate processes to speed up incremental updates! 在两个独立进程中同时运行，加速增量更新！ Incremental update logic: - Read existing file, get latest date - Only fetch data newer than latest date (add to file, don't overwrite) - If file doesn't exist, fetch full 800 points - Process max 100 stocks per run (incremental) / 800 (new file) - Update database timestamp after successful fetch """ import os import sys import sqlite3 import requests import json from datetime import datetime # Fix encoding for Windows console if sys.stdout.encoding.lower() != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') # Configuration DB_PATH = "D:\\xistock\\stock.db" DATA_DIR = "D:\\xistock\\week" BASE_URL = "https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get" DATA_POINTS_FULL = 800 # When file doesn't exist MAX_STOCKS_PER_RUN_INCREMENTAL = 100 # Max stocks to process per run when incremental update FREQ = "week" BATCH_SIZE = 3 # Process 2-3 stocks per batch to avoid long runtime def get_db_connection(): """Get database connection""" if not os.path.exists(DB_PATH): print(f"- Database not found: {DB_PATH}") print(" Please run init_db.py and fetch_stocks.py first!") exit(1) return sqlite3.connect(DB_PATH) def get_latest_date_from_file(filepath): """ Read existing file and get the latest date Returns None if file doesn't exist or empty """ if not os.path.exists(filepath): return None try: with open(filepath, 'r', encoding='utf-8') as f: # Skip header next(f) latest_date = None # Read all lines to get the last date for line in f: line = line.strip() if not line: continue date = line.split(',')[0] latest_date = date return latest_date except Exception as e: print(f" Warning: Cannot read existing file: {e}") return None def get_stocks_to_fetch(order): """ Get list of stocks that need to be fetched (incremental) - Selects stocks that have never been fetched or haven't been fetched recently - order: 'asc' or 'desc' - split into two partitions - Returns up to MAX_STOCKS_PER_RUN_INCREMENTAL stocks """ conn = get_db_connection() cursor = conn.cursor() # Get all active stocks ordered by code cursor.execute(""" SELECT code, name, market, week_get FROM stocks WHERE status = 'active' ORDER BY code """) all_stocks = cursor.fetchall() total = len(all_stocks) half = total // 2 # Split into two partitions if order == 'asc': partition_stocks = all_stocks[:half] elif order == 'desc': all_stocks.reverse() partition_stocks = all_stocks[:half] else: print(f"- Invalid order: {order}. Use 'asc' or 'desc'") exit(1) # Filter stocks that need update (any stock that hasn't been fetched today) today = datetime.now().strftime('%Y-%m-%d') need_update = [] for stock in partition_stocks: if len(stock) == 3: code, name, market = stock week_get = None else: code, name, market, week_get = stock # If never fetched or last fetched not today, needs update if week_get is None or not week_get.startswith(today): need_update.append((code, name, market)) # Process all need update today, large safety limit to prevent infinite run # MAX_STOCKS_PER_RUN_INCREMENTAL is original limit, now we allow 10x that if len(need_update) > MAX_STOCKS_PER_RUN_INCREMENTAL * 10: need_update = need_update[:MAX_STOCKS_PER_RUN_INCREMENTAL * 10] print(f"Partition {order}:") print(f" Total in partition: {len(partition_stocks)}") print(f" Need update today: {len(need_update)}") print(f" Will process all {len(need_update)} stocks (safety limit: {MAX_STOCKS_PER_RUN_INCREMENTAL * 10})") print(f" Will process: {len(need_update)} (max {MAX_STOCKS_PER_RUN_INCREMENTAL})") conn.close() return need_update def format_stock_code(code): """Format stock code for Tencent Finance API (sh600001, sz000001)""" if code.startswith('60'): return f"sh{code}" else: return f"sz{code}" def fetch_stock_data(stock_code, stock_name): """Fetch weekly K-line data from Tencent Finance API""" try: tencent_code = format_stock_code(stock_code) params = f"{tencent_code},{FREQ},,,{DATA_POINTS_FULL},qfq" url = f"{BASE_URL}?_var=kline_weekqfq¶m={params}" response = requests.get(url, timeout=30) response.encoding = 'utf-8' if response.status_code != 200: print(f"- {stock_code} {stock_name}: HTTP {response.status_code}") return None # Response is in format: kline_weekqfq = {...}; content = response.text if '=' in content: json_str = content.split('=', 1)[1].rstrip(';') else: json_str = content data = json.loads(json_str) # Data structure: data -> tencent_code -> qfqweek if 'data' not in data or tencent_code not in data['data']: print(f"- {stock_code} {stock_name}: No data found") return None stock_data = data['data'][tencent_code] if 'qfqweek' not in stock_data: print(f"- {stock_code} {stock_name}: No qfqweek found") return None return stock_data['qfqweek'] except Exception as e: print(f"- {stock_code} {stock_name}: Error fetching - {str(e)}") return None def process_data(data): """ Process raw data, extract date, open, close, high, low and calculate change 返回格式列表: [date, open, close, high, low, change] """ processed = [] # Data format from Tencent: [date, open, close, high, low, volume, ...] for bar in data: if len(bar) >= 5: date = bar[0] open_p = float(bar[1]) close = float(bar[2]) high = float(bar[3]) low = float(bar[4]) processed.append([date, open_p, close, high, low]) # Calculate change (涨跌幅) for i in range(len(processed)): if i == 0: change = 0.0 else: prev_close = processed[i-1][2] if prev_close == 0: change = 0.0 else: change = ((processed[i][2] - prev_close) / prev_close) * 100 processed[i].append(round(change, 3)) return processed def get_data_points_to_fetch(processed_new, latest_date): """ Filter new data points that are after latest_date in existing file If latest_date is None, return all """ if latest_date is None: return processed_new, DATA_POINTS_FULL # Find all data points after latest_date new_data = [] for item in processed_new: date = item[0] if date > latest_date: new_data.append(item) return new_data, len(processed_new) def append_to_file(stock_code, stock_name, new_data, latest_date): """ Append new data points to existing file, don't overwrite If file doesn't exist, create new with header Format: date,open,close,high,low,change Filename: 名称_代码.txt """ # Create directory if not exists if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) # Sanitize filename - remove all invalid Windows filename characters invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = stock_name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{stock_code}.txt" filepath = os.path.join(DATA_DIR, filename) # Filter new data (only data after latest_date) data_to_write, total_fetched = get_data_points_to_fetch(new_data, latest_date) if not data_to_write: # No new data to add return filepath, 0 if latest_date is None: # Create new file with open(filepath, 'w', encoding='utf-8') as f: f.write("date,open,close,high,low,change_pct\n") for item in data_to_write: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) else: # Append to existing file with open(filepath, 'a', encoding='utf-8') as f: for item in data_to_write: date, open_p, close, high, low, change = item line = f"{date},{open_p:.2f},{close:.2f},{high:.2f},{low:.2f},{change:.3f}\n" f.write(line) return filepath, len(data_to_write) def update_database_last_fetch(stock_code): """Update the week_get timestamp in database to current date""" conn = get_db_connection() cursor = conn.cursor() current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') cursor.execute(""" UPDATE stocks SET week_get = ? WHERE code = ? """, (current_time, stock_code)) conn.commit() conn.close() def fetch_all_stocks(order): """Fetch data incrementally for stocks in this partition - process {BATCH_SIZE} per run""" stocks = get_stocks_to_fetch(order) total_need_update = len(stocks) print("=" * 70) print(f"A-Share-Get Weekly Data Fetcher - Incremental Parallel ({order})") print("A股数据获取 - 周线增量更新（并行）") print("=" * 70) print(f"Data directory: {DATA_DIR}") print(f"Processing {BATCH_SIZE} stocks per run (max {MAX_STOCKS_PER_RUN_INCREMENTAL} total)") print("-" * 70) success_total = 0 failed_total = 0 new_points_total = 0 # Process in batches of BATCH_SIZE from itertools import islice def batch_generator(iterable, batch_size): iterator = iter(iterable) for first in iterator: yield [first] + list(islice(iterator, batch_size - 1)) for batch_idx, batch in enumerate(batch_generator(stocks, BATCH_SIZE), 1): print(f"\n>> Batch {batch_idx}, processing {len(batch)} stocks:") for i, (code, name, market) in enumerate(batch, 1): global_i = (batch_idx - 1) * BATCH_SIZE + i print(f" [{global_i}/{total_need_update}] {code} {name}: ", end=' ') sys.stdout.flush() # Get existing file's latest date invalid_chars = ['\\', '/', ':', '*', '?', '"', '<', '>', '|'] safe_name = name for c in invalid_chars: safe_name = safe_name.replace(c, '_') filename = f"{safe_name}_{code}.txt" filepath = os.path.join(DATA_DIR, filename) latest_date = get_latest_date_from_file(filepath) if latest_date: print(f"latest={latest_date}, fetching new data...", end=' ') else: print("new file, fetching full data...", end=' ') sys.stdout.flush() # Fetch data raw_data = fetch_stock_data(code, name) if raw_data is None: print("FAILED") failed_total += 1 continue # Process data processed = process_data(raw_data) if not processed: print("NO DATA") failed_total += 1 continue # Append new data to file (only add missing) filepath, new_count = append_to_file(code, name, processed, latest_date) if new_count == 0: print("no new data") # Still update database to mark as updated today update_database_last_fetch(code) continue # Update database update_database_last_fetch(code) print(f"OK added {new_count} new points") success_total += 1 new_points_total += new_count # Add delay to avoid hitting rate limit import time time.sleep(1.5) # Final summary print("-" * 70) print("增量更新完成！") print("-" * 70) print(f"Partition ({order}):") print(f" - Need update: {total_need_update}") print(f" - Successfully processed: {success_total}") print(f" - Failed: {failed_total}") print(f" - New data points added: {new_points_total}") print(f"Data directory: {DATA_DIR}") print("-" * 70) print("Database updated with last fetch timestamps") print("=" * 70) if __name__ == "__main__": if len(sys.argv) != 2: print("Usage:") print(" python scripts/week_parallel.py asc # First half (ascending)") print(" python scripts/week_parallel.py desc # Second half (descending)") print("\nRun both in separate processes for parallel incremental fetching!") exit(1) order = sys.argv[1].lower() try: fetch_all_stocks(order) except KeyboardInterrupt: print("\n\n- Interrupted by user") exit(1) except Exception as e: print(f"\n- Error: {e}") import traceback traceback.print_exc() exit(1)

A股数据获取 A specialized data collection tool for Chinese A-share market

Comments (0)