From 28d1606947056c8215002e9bd2b31e66231120bb Mon Sep 17 00:00:00 2001 From: claude_dev Date: Thu, 23 Jul 2026 09:04:01 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20akshare=20=E4=BD=8E=E9=A2=91=205?= =?UTF-8?q?=20schtask=20wrapper+register+verify(spec=20=C2=A714.5=20A+B/C)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 5 wrapper(ak_eod daily19:00估值+财务摘要/ak_quarter财报季三表+预告快报/ ak_events daily19:30龙虎榜+大宗+两融+解禁 per-date当日/ak_stock weekly周六 北向+股本+十大股东/index_monthly 16号成份股3源+migrate+merge)+register_akshare _schtasks(/m APR,MAY,SEP,NOV财报季)+verify_akshare_e2e; 均unset proxy+夜间避封IP --- scripts/data_platform/ak_eod_wrapper.ps1 | 12 +++ scripts/data_platform/ak_events_wrapper.ps1 | 13 ++++ scripts/data_platform/ak_quarter_wrapper.ps1 | 12 +++ scripts/data_platform/ak_stock_wrapper.ps1 | 12 +++ .../data_platform/index_monthly_wrapper.ps1 | 32 ++++++++ .../register_akshare_schtasks.ps1 | 15 ++++ scripts/data_platform/verify_akshare_e2e.py | 75 +++++++++++++++++++ 7 files changed, 171 insertions(+) create mode 100644 scripts/data_platform/ak_eod_wrapper.ps1 create mode 100644 scripts/data_platform/ak_events_wrapper.ps1 create mode 100644 scripts/data_platform/ak_quarter_wrapper.ps1 create mode 100644 scripts/data_platform/ak_stock_wrapper.ps1 create mode 100644 scripts/data_platform/index_monthly_wrapper.ps1 create mode 100644 scripts/data_platform/register_akshare_schtasks.ps1 create mode 100644 scripts/data_platform/verify_akshare_e2e.py diff --git a/scripts/data_platform/ak_eod_wrapper.ps1 b/scripts/data_platform/ak_eod_wrapper.ps1 new file mode 100644 index 0000000..96a4f07 --- /dev/null +++ b/scripts/data_platform/ak_eod_wrapper.ps1 @@ -0,0 +1,12 @@ +# ak_eod_wrapper.ps1 — sanguo-ak-eod schtask wrapper (daily 19:00 akshare 估值+财务摘要 --force) +# per-stock marker 是 symbol 级, --force 才能刷新当日新数据; 5500股×2×0.8s≈2.5h 夜间 +$env:http_proxy = '' +$env:https_proxy = '' +$env:all_proxy = '' +Set-Location C:\sanguo_vnpy_v2 +$ts = Get-Date -Format 'yyyyMMdd_HHmmss' +$logDir = 'C:\sanguo_vnpy_v2\data\migration_logs' +if (-not (Test-Path $logDir)) { New-Item -ItemType Directory -Path $logDir -Force | Out-Null } +$log = Join-Path $logDir "ak_eod_$ts.txt" +C:\Python310\python.exe -X utf8 C:\sanguo_vnpy_v2\scripts\data_platform\akshare_static_download.py --types valuation,financial_abstract --force *>> $log +exit $LASTEXITCODE diff --git a/scripts/data_platform/ak_events_wrapper.ps1 b/scripts/data_platform/ak_events_wrapper.ps1 new file mode 100644 index 0000000..8d8dfcd --- /dev/null +++ b/scripts/data_platform/ak_events_wrapper.ps1 @@ -0,0 +1,13 @@ +# ak_events_wrapper.ps1 — sanguo-ak-events schtask wrapper (daily 19:30 akshare 事件类 per-date 当日) +# dragon_tiger/block_trade/margin_sse/restricted --start today --end today; per-date 每类1 unit 快 +$env:http_proxy = '' +$env:https_proxy = '' +$env:all_proxy = '' +Set-Location C:\sanguo_vnpy_v2 +$today = Get-Date -Format yyyyMMdd +$ts = Get-Date -Format 'yyyyMMdd_HHmmss' +$logDir = 'C:\sanguo_vnpy_v2\data\migration_logs' +if (-not (Test-Path $logDir)) { New-Item -ItemType Directory -Path $logDir -Force | Out-Null } +$log = Join-Path $logDir "ak_events_$ts.txt" +C:\Python310\python.exe -X utf8 C:\sanguo_vnpy_v2\scripts\data_platform\akshare_static_download.py --types dragon_tiger,block_trade,margin_sse,restricted --start $today --end $today *>> $log +exit $LASTEXITCODE diff --git a/scripts/data_platform/ak_quarter_wrapper.ps1 b/scripts/data_platform/ak_quarter_wrapper.ps1 new file mode 100644 index 0000000..f780490 --- /dev/null +++ b/scripts/data_platform/ak_quarter_wrapper.ps1 @@ -0,0 +1,12 @@ +# ak_quarter_wrapper.ps1 — sanguo-ak-quarter schtask wrapper (财报季 APR/MAY/SEP/NOV 02:00 akshare 三表+预告/快报 --force) +# balance+income+cashflow+forecast+express; per-stock 全量慢(5500×5), 财报季月夜间跑 +$env:http_proxy = '' +$env:https_proxy = '' +$env:all_proxy = '' +Set-Location C:\sanguo_vnpy_v2 +$ts = Get-Date -Format 'yyyyMMdd_HHmmss' +$logDir = 'C:\sanguo_vnpy_v2\data\migration_logs' +if (-not (Test-Path $logDir)) { New-Item -ItemType Directory -Path $logDir -Force | Out-Null } +$log = Join-Path $logDir "ak_quarter_$ts.txt" +C:\Python310\python.exe -X utf8 C:\sanguo_vnpy_v2\scripts\data_platform\akshare_static_download.py --types balance,income,cashflow,forecast,express --force *>> $log +exit $LASTEXITCODE diff --git a/scripts/data_platform/ak_stock_wrapper.ps1 b/scripts/data_platform/ak_stock_wrapper.ps1 new file mode 100644 index 0000000..50c1129 --- /dev/null +++ b/scripts/data_platform/ak_stock_wrapper.ps1 @@ -0,0 +1,12 @@ +# ak_stock_wrapper.ps1 — sanguo-ak-stock schtask wrapper (weekly 周六03:00 akshare per-stock 慢) +# northbound+share_capital+top_holders --force; per-stock 全量慢, 周末夜间 +$env:http_proxy = '' +$env:https_proxy = '' +$env:all_proxy = '' +Set-Location C:\sanguo_vnpy_v2 +$ts = Get-Date -Format 'yyyyMMdd_HHmmss' +$logDir = 'C:\sanguo_vnpy_v2\data\migration_logs' +if (-not (Test-Path $logDir)) { New-Item -ItemType Directory -Path $logDir -Force | Out-Null } +$log = Join-Path $logDir "ak_stock_$ts.txt" +C:\Python310\python.exe -X utf8 C:\sanguo_vnpy_v2\scripts\data_platform\akshare_static_download.py --types northbound,share_capital,top_holders --force *>> $log +exit $LASTEXITCODE diff --git a/scripts/data_platform/index_monthly_wrapper.ps1 b/scripts/data_platform/index_monthly_wrapper.ps1 new file mode 100644 index 0000000..c680573 --- /dev/null +++ b/scripts/data_platform/index_monthly_wrapper.ps1 @@ -0,0 +1,32 @@ +# index_monthly_wrapper.ps1 — sanguo-index schtask wrapper (monthly 16号 19:50 成份股刷新) +# 1. index_const_hist_download 深证(A adjust_cni)+中证(B snapshot) akshare parquet 刷新(成份股随时间变) +# 2. migrate_constituent 重建 staging(从 bs_index_constituent_old baostock 全集 + 深证/中证 parquet) +# 3. merge_constituent staging -> constituent_unified (幂等 DROP+CREATE) +# 不碰 baostock online(全集已在 _old, migrate 读它); 不跑 C 新浪(baostock 权威) +$env:http_proxy = '' +$env:https_proxy = '' +$env:all_proxy = '' +$env:PYTHONIOENCODING = 'utf-8' +$env:INDEX_HIST_OUT_DIR = 'C:\sanguo_vnpy_v2\data\index_const_hist' +Set-Location C:\sanguo_vnpy_v2 +$ts = Get-Date -Format 'yyyyMMdd_HHmmss' +$logDir = 'C:\sanguo_vnpy_v2\data\migration_logs' +if (-not (Test-Path $logDir)) { New-Item -ItemType Directory -Path $logDir -Force | Out-Null } +$log = Join-Path $logDir "index_monthly_$ts.txt" +$py = 'C:\Python310\python.exe' +$dir = 'C:\sanguo_vnpy_v2\scripts\data_platform' + +"[$ts] STEP1 index_const_hist_download --parts A B (深证 adjust + 中证 snapshot)" | Out-File $log -Encoding UTF8 +& $py -X utf8 "$dir\index_const_hist_download.py" --parts A B *>> $log +"[$ts] STEP1 done exit=$LASTEXITCODE" | Out-File $log -Append -Encoding UTF8 + +"[$ts] STEP2 migrate_constituent (rebuild staging from _old + parquet)" | Out-File $log -Append -Encoding UTF8 +& $py -X utf8 "$dir\migrate_constituent.py" *>> $log +"[$ts] STEP2 done exit=$LASTEXITCODE" | Out-File $log -Append -Encoding UTF8 + +"[$ts] STEP3 merge_constituent (staging -> constituent_unified, idempotent)" | Out-File $log -Append -Encoding UTF8 +& $py -X utf8 "$dir\merge_constituent.py" *>> $log +$final = $LASTEXITCODE +"[$ts] STEP3 done exit=$final" | Out-File $log -Append -Encoding UTF8 + +exit $final diff --git a/scripts/data_platform/register_akshare_schtasks.ps1 b/scripts/data_platform/register_akshare_schtasks.ps1 new file mode 100644 index 0000000..001d250 --- /dev/null +++ b/scripts/data_platform/register_akshare_schtasks.ps1 @@ -0,0 +1,15 @@ +# register_akshare_schtasks.ps1 — 注册 akshare 低频 5 schtask (spec §14.5 A+B+C) +# akshare per-stock 全量慢+东财限流 → 夜间/周末/财报季; akshare_static_download 内置断路器(连30 failed exit)+0.8s sleep +# 时间错开 bs-eod(18:05)/xt-eod(18:40) 之后, 无并发 +$ws = 'C:\sanguo_vnpy_v2\scripts\data_platform' +schtasks /create /tn sanguo-ak-eod /tr "powershell -ExecutionPolicy Bypass -File $ws\ak_eod_wrapper.ps1" /sc daily /st 19:00 /ru SYSTEM /rl HIGHEST /f +schtasks /create /tn sanguo-ak-events /tr "powershell -ExecutionPolicy Bypass -File $ws\ak_events_wrapper.ps1" /sc daily /st 19:30 /ru SYSTEM /rl HIGHEST /f +schtasks /create /tn sanguo-ak-stock /tr "powershell -ExecutionPolicy Bypass -File $ws\ak_stock_wrapper.ps1" /sc weekly /d SAT /st 03:00 /ru SYSTEM /rl HIGHEST /f +schtasks /create /tn sanguo-ak-quarter /tr "powershell -ExecutionPolicy Bypass -File $ws\ak_quarter_wrapper.ps1" /sc monthly /m APR,MAY,SEP,NOV /st 02:00 /ru SYSTEM /rl HIGHEST /f +schtasks /create /tn sanguo-index /tr "powershell -ExecutionPolicy Bypass -File $ws\index_monthly_wrapper.ps1" /sc monthly /d 16 /st 19:50 /ru SYSTEM /rl HIGHEST /f +Write-Output "=== VERIFY ===" +foreach ($t in @('sanguo-ak-eod','sanguo-ak-events','sanguo-ak-stock','sanguo-ak-quarter','sanguo-index')) { + $q = schtasks /query /tn $t /fo list 2>&1 | Out-String + $status = ($q -split "`n" | Where-Object { $_ -match 'Status' } | Select-Object -First 1).Trim() + Write-Output (" {0,-22} {1}" -f $t, $status) +} diff --git a/scripts/data_platform/verify_akshare_e2e.py b/scripts/data_platform/verify_akshare_e2e.py new file mode 100644 index 0000000..f125670 --- /dev/null +++ b/scripts/data_platform/verify_akshare_e2e.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""verify_akshare_e2e.py — akshare 低频 schtask E2E 验证(VPS 跑)。 +验: static/ 各表文件数+最新mtime / events/ per-date 类 parquet / constituent_unified 行数+3source / bs_index_constituent_old 完好。 +stdout 行缓冲(ssh 非交互块缓冲兼容)。 +""" +import datetime +import sqlite3 +import sys +from pathlib import Path + +sys.stdout.reconfigure(line_buffering=True) + +BASE = Path(r"C:\sanguo_vnpy_v2\data") +ok, warn = [], [] + +# 1. static/ 各表文件数 + 最新 mtime +print("===== static/ 各表 =====") +for sub in ["valuation", "financial_abstract", "balance", "income", "cashflow", + "northbound", "share_capital", "top_holders"]: + d = BASE / "static" / sub + if not d.exists(): + warn.append(f"static/{sub} MISSING") + print(f" {sub}: MISSING") + continue + fs = list(d.glob("*.parquet")) + if not fs: + warn.append(f"static/{sub} 0 parquet") + print(f" {sub}: 0") + continue + mt = max(f.stat().st_mtime for f in fs) + latest = datetime.datetime.fromtimestamp(mt).strftime("%Y-%m-%d %H:%M") + print(f" {sub}: {len(fs)} parquet, latest={latest}") + ok.append(f"static/{sub}") + +# 2. events/ per-date 类 +print("\n===== events/ per-date 类 =====") +for sub in ["dragon_tiger", "block_trade", "margin_sse", "restricted"]: + d = BASE / "events" / sub + if not d.exists(): + warn.append(f"events/{sub} MISSING(ak-events 未跑或当日无数据)") + print(f" {sub}: MISSING") + continue + fs = list(d.glob("*.parquet")) + print(f" {sub}: {len(fs)} parquet") + if fs: + ok.append(f"events/{sub}") + +# 3. constituent_unified + _old +print("\n===== constituent_unified + _old =====") +c = sqlite3.connect(str(BASE / "quant_trading.db"), timeout=60) +try: + c.execute("PRAGMA busy_timeout = 60000") + n = c.execute("SELECT COUNT(*) FROM constituent_unified").fetchone()[0] + src = c.execute("SELECT source, COUNT(*) FROM constituent_unified GROUP BY source").fetchall() + print(f" constituent_unified: {n} rows, sources={src}") + if n >= 7110 and len(src) >= 3: + ok.append("constituent_unified>=7110+3source") + else: + warn.append(f"constituent_unified {n}<7110 或 source<3 {src}") + try: + nold = c.execute("SELECT COUNT(*) FROM bs_index_constituent_old").fetchone()[0] + print(f" bs_index_constituent_old: {nold} rows (baostock 历史时点完好)") + if nold > 0: + ok.append("_old intact") + except Exception as e: + warn.append(f"_old 查询失败 {e}") +finally: + c.close() + +print("\n===== SUMMARY =====") +print(f"OK: {len(ok)} WARN: {len(warn)}") +for w in warn: + print(f" WARN {w}") +print("VERIFY DONE")