Compare commits

...

2 Commits

Author SHA1 Message Date
fish 8031f27f85 新增同步 ext_data(概念/行业) 和本地化财务数据展示
- 将 ext_data 加入 SYNCABLE_PARTS,概念/行业分类数据可从 local 同步到 serve
- 去掉 serve 财务分析接口的 Cap.FINANCIAL 能力门控,依赖本地 parquet 文件来展示数据
- serve 财务分析前端在无 API Key 但有同步数据时正常展示,隐藏同步按钮

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-07-06 10:53:00 +08:00
fish 274b90eef6 新增 local→serve 数据同步功能
Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-06 10:27:00 +08:00
13 changed files with 785 additions and 77 deletions
+201
View File
@@ -0,0 +1,201 @@
"""数据同步 — 将本地 data/ 打包推送到 serve 端。"""
from __future__ import annotations
import json
import logging
import tarfile
import tempfile
from datetime import datetime
from io import BytesIO
from pathlib import Path
import httpx
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel
from app.config import settings
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/data/sync", tags=["data-sync"])
# 允许同步的子目录 — 覆盖 serve 功能面板所需的数据
SYNCABLE_PARTS = [
"kline_daily",
"kline_daily_enriched",
"kline_index_daily",
"kline_index_enriched",
"kline_etf_daily",
"kline_etf_enriched",
"kline_etf_minute",
"kline_minute",
"adj_factor",
"adj_factor_etf",
"instruments",
"instruments_index",
"instruments_etf",
"financials",
"pools",
"ext_data",
]
def _part_size_info(data_dir: Path, part: str) -> dict:
"""获取指定子目录的文件数和大小。"""
part_dir = data_dir / part
if not part_dir.exists():
return {"file_count": 0, "size_bytes": 0, "last_modified": None}
files = list(part_dir.rglob("*.parquet"))
if not files:
return {"file_count": 0, "size_bytes": 0, "last_modified": None}
total = sum(f.stat().st_size for f in files)
mtimes = [f.stat().st_mtime for f in files]
last_ts = datetime.fromtimestamp(max(mtimes)).isoformat() if mtimes else None
return {"file_count": len(files), "size_bytes": total, "last_modified": last_ts}
def _pack_parts(data_dir: Path, parts: list[str]) -> bytes:
"""将指定子目录打包为 tar.gz 字节流。"""
buf = BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
for part in parts:
part_dir = data_dir / part
if not part_dir.exists():
continue
for fpath in sorted(part_dir.rglob("*.parquet")):
rel = fpath.relative_to(data_dir)
tar.add(fpath, arcname=str(rel))
buf.seek(0)
return buf.getvalue()
class SyncConfig(BaseModel):
serve_url: str = ""
sync_key: str = ""
enable_auto: bool = False
interval_minutes: int = 0
# ── 存储配置用的 JSON 文件 ──────────────────────────────────
CONFIG_FILE = "sync_config.json"
def _config_path(data_dir: Path) -> Path:
return data_dir / "user_data" / CONFIG_FILE
def _load_config(data_dir: Path) -> SyncConfig:
cp = _config_path(data_dir)
if cp.exists():
try:
return SyncConfig(**json.loads(cp.read_text()))
except Exception:
pass
return SyncConfig()
def _save_config(data_dir: Path, cfg: SyncConfig) -> None:
cp = _config_path(data_dir)
cp.parent.mkdir(parents=True, exist_ok=True)
cp.write_text(cfg.model_dump_json(indent=2))
# ── 路由 ──────────────────────────────────────────────────
@router.get("/status")
def sync_status(request: Request):
"""各数据目录的本地状态 + 同步配置。"""
data_dir: Path = request.app.state.datastore.data_dir
cfg = _load_config(data_dir)
parts = {}
for p in SYNCABLE_PARTS:
parts[p] = _part_size_info(data_dir, p)
return {
"parts": parts,
"config": cfg.model_dump(),
"last_sync": cfg.model_dump().get("_last_sync_time"),
}
class ConfigReq(BaseModel):
serve_url: str
sync_key: str
enable_auto: bool = False
interval_minutes: int = 0
@router.post("/config")
def update_sync_config(request: Request, body: ConfigReq):
"""更新同步配置。"""
data_dir: Path = request.app.state.datastore.data_dir
cfg = SyncConfig(
serve_url=body.serve_url.rstrip("/"),
sync_key=body.sync_key,
enable_auto=body.enable_auto,
interval_minutes=body.interval_minutes,
)
_save_config(data_dir, cfg)
return {"ok": True}
class PushReq(BaseModel):
parts: list[str]
@router.post("/push")
async def push_sync(request: Request, body: PushReq):
"""打包指定 parts 推送到 serve 端。"""
data_dir: Path = request.app.state.datastore.data_dir
cfg = _load_config(data_dir)
if not cfg.serve_url:
raise HTTPException(status_code=400, detail="未配置 serve_url,请先在同步设置中配置")
requested = [p for p in body.parts if p in SYNCABLE_PARTS]
if not requested:
raise HTTPException(status_code=400, detail="未指定有效的同步目录")
# 检查有没有数据可同步
available = [p for p in requested if (data_dir / p).exists()]
if not available:
raise HTTPException(status_code=400, detail="所选目录均无数据可同步")
# 打包
tarball = _pack_parts(data_dir, available)
size_bytes = len(tarball)
logger.info("packed %d parts -> %d bytes", len(available), size_bytes)
# 推送
upload_url = f"{cfg.serve_url}/api/data/sync/upload"
headers = {"X-Sync-Key": cfg.sync_key} if cfg.sync_key else {}
try:
async with httpx.AsyncClient(timeout=600) as client:
resp = await client.post(
upload_url,
files={"file": ("sync.tar.gz", tarball, "application/gzip")},
data={"parts": ",".join(available)},
headers=headers,
)
if resp.status_code != 200:
detail = resp.text[:500]
raise HTTPException(
status_code=502,
detail=f"serve 返回异常 ({resp.status_code}): {detail}",
)
result = resp.json()
except httpx.RequestError as exc:
raise HTTPException(status_code=502, detail=f"连接 serve 失败: {exc}") from exc
# 记录上次同步时间
cfg_dict = cfg.model_dump()
cfg_dict["_last_sync_time"] = datetime.now().isoformat()
cfg = SyncConfig(**cfg_dict)
_save_config(data_dir, cfg)
logger.info("sync pushed: parts=%s files=%d", available, result.get("file_count", 0))
return {
"ok": True,
"parts_sent": available,
"size_bytes": size_bytes,
"serve_response": result,
}
+4
View File
@@ -95,6 +95,10 @@ class Settings(BaseSettings):
log_level: str = "INFO"
backtest_range_guard: bool = False
# Sync — 数据同步到 serve
sync_serve_url: str = ""
sync_key: str = ""
# Auth — 首次启动时预置访问密码(明文, 仅用于初始化, 详见 services/auth.bootstrap_from_env)
# 公网服务器部署时免去 SSH 端口转发设密码的麻烦。写入 auth.json(哈希)后即不再读取。
auth_password: str = ""
+3 -2
View File
@@ -11,7 +11,7 @@ from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from app import __version__
from app.api import analysis, auth as auth_api, backtest, data, ext_data, financials, indices, intraday, kline, market_recap, monitor_rules, alerts, overview, pipeline, rps, screener, settings as settings_api, signals, stock_analysis, strategy, watchlist
from app.api import analysis, auth as auth_api, backtest, data, data_sync, ext_data, financials, indices, intraday, kline, market_recap, monitor_rules, alerts, overview, pipeline, rps, screener, settings as settings_api, signals, stock_analysis, strategy, watchlist
from app.api.routes import router as core_router
from app.config import settings
from app.jobs import daily_pipeline
@@ -207,7 +207,7 @@ app.add_middleware(
# 2. 未设密码 + 公网 → 拒绝(403, 防裸奔也防抢占; 引导本机设密码)
# 3. 已设密码 → 检查 session, 无效则 401(前端跳登录)
# 白名单: /api/auth/* (设密码/登录本身)、/health 等探活。
_AUTH_WHITELIST_PREFIX = ("/api/auth/",)
_AUTH_WHITELIST_PREFIX = ("/api/auth/", "/api/data/sync/")
_AUTH_WHITELIST_EXACT = ("/health", "/api/health", "/openapi.json", "/docs", "/redoc")
@@ -257,6 +257,7 @@ app.include_router(overview.router)
app.include_router(analysis.router)
app.include_router(pipeline.router)
app.include_router(data.router)
app.include_router(data_sync.router)
app.include_router(ext_data.router)
app.include_router(financials.router)
app.include_router(stock_analysis.router)
+1 -1
View File
@@ -7,7 +7,7 @@ services:
dockerfile: Dockerfile
args:
BACKEND_EXTRAS: ${BACKEND_EXTRAS:-}
container_name: TickFlow_Stock_Panel
container_name: TickFlow_Local
ports:
- "${PORT:-3018}:3018"
env_file:
+2
View File
@@ -44,6 +44,7 @@ import {
BookOpenCheck,
ExternalLink,
X,
Upload,
} from 'lucide-react'
import { Logo } from './Logo'
import { api, type IndexQuote } from '@/lib/api'
@@ -78,6 +79,7 @@ const nav = [
{ to: '/indices', label: '指数', icon: BarChart3 },
{ to: '/trading', label: '交易', icon: Cable },
{ to: '/data', label: '数据', icon: Database },
{ to: '/sync', label: '数据同步', icon: Upload },
] as const
function fmtIndexValue(v: number | null | undefined) {
+44
View File
@@ -1752,6 +1752,14 @@ export const api = {
method: 'POST',
body: JSON.stringify({ strategy_id: strategyId, code }),
}),
// ===== Sync =====
syncStatus: (): Promise<SyncStatusResponse> =>
request('/api/data/sync/status'),
syncConfig: (body: SyncConfigReq): Promise<{ ok: boolean }> =>
request('/api/data/sync/config', { method: 'POST', body: JSON.stringify(body) }),
syncPush: (parts: string[]): Promise<SyncPushResponse> =>
request('/api/data/sync/push', { method: 'POST', body: JSON.stringify({ parts }) }),
}
// ===== Pipeline =====
@@ -1931,3 +1939,39 @@ export interface AnalysisMenu {
updated_at?: string | null
builtin?: boolean
}
// ===== Sync =====
export interface SyncPartInfo {
file_count: number
size_bytes: number
last_modified: string | null
}
export interface SyncStatusResponse {
parts: Record<string, SyncPartInfo>
config: {
serve_url: string
sync_key: string
enable_auto: boolean
interval_minutes: number
}
last_sync?: string | null
}
export interface SyncPushResponse {
ok: boolean
parts_sent: string[]
size_bytes: number
serve_response: {
ok: boolean
parts: string[]
file_count: number
}
}
export interface SyncConfigReq {
serve_url: string
sync_key: string
enable_auto?: boolean
interval_minutes?: number
}
+310
View File
@@ -0,0 +1,310 @@
import { useCallback, useEffect, useState } from 'react'
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
import {
CheckCircle2,
Cloud,
CloudOff,
Database,
HardDrive,
Loader2,
RefreshCw,
Save,
Settings,
Upload,
} from 'lucide-react'
import { toast } from '@/components/Toast'
import { PageHeader } from '@/components/PageHeader'
import { api, type SyncStatusResponse, type SyncPushResponse } from '@/lib/api'
import { cn } from '@/lib/cn'
type PartEntry = {
key: string
label: string
}
const ALL_PARTS: PartEntry[] = [
{ key: 'kline_daily', label: '日 K 线' },
{ key: 'kline_daily_enriched', label: '日 K 增强' },
{ key: 'kline_index_daily', label: '指数日 K' },
{ key: 'kline_index_enriched', label: '指数日 K 增强' },
{ key: 'kline_etf_daily', label: 'ETF 日 K' },
{ key: 'kline_etf_enriched', label: 'ETF 日 K 增强' },
{ key: 'kline_etf_minute', label: 'ETF 分钟 K' },
{ key: 'kline_minute', label: '分钟 K' },
{ key: 'adj_factor', label: '复权因子' },
{ key: 'adj_factor_etf', label: 'ETF 复权因子' },
{ key: 'instruments', label: '股票列表' },
{ key: 'instruments_index', label: '指数列表' },
{ key: 'instruments_etf', label: 'ETF 列表' },
{ key: 'financials', label: '财务数据' },
{ key: 'pools', label: '板块池' },
]
function fmtSize(bytes: number): string {
if (bytes === 0) return '—'
if (bytes < 1024) return `${bytes} B`
if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`
return `${(bytes / (1024 * 1024)).toFixed(1)} MB`
}
export function Sync() {
const qc = useQueryClient()
const [showSettings, setShowSettings] = useState(false)
const [serveUrl, setServeUrl] = useState('')
const [syncKey, setSyncKey] = useState('')
const [selected, setSelected] = useState<Set<string>>(new Set(ALL_PARTS.map(p => p.key)))
// Status
const { data: status, isLoading, refetch } = useQuery<SyncStatusResponse>({
queryKey: ['sync-status'],
queryFn: api.syncStatus,
refetchInterval: 30000,
})
// Init form from server config
useEffect(() => {
if (status?.config) {
setServeUrl(status.config.serve_url)
setSyncKey(status.config.sync_key)
}
}, [status?.config])
// Save config
const saveCfg = useMutation<{ ok: boolean }, Error, void>({
mutationFn: () => api.syncConfig({ serve_url: serveUrl, sync_key: syncKey }),
onSuccess: () => {
toast('同步配置已保存', 'success')
qc.invalidateQueries({ queryKey: ['sync-status'] })
setShowSettings(false)
},
})
// Push sync
const push = useMutation<SyncPushResponse, Error, void>({
mutationFn: () => api.syncPush([...selected]),
onSuccess: (res) => {
if (res.ok) {
toast(
`同步完成: ${res.parts_sent.length} 个目录, ${res.serve_response.file_count} 个文件`,
'success',
)
qc.invalidateQueries({ queryKey: ['sync-status'] })
}
},
})
const togglePart = useCallback((key: string) => {
setSelected(prev => {
const next = new Set(prev)
if (next.has(key)) next.delete(key)
else next.add(key)
return next
})
}, [])
const selectAll = () => setSelected(new Set(ALL_PARTS.map(p => p.key)))
const selectNone = () => setSelected(new Set())
return (
<div className="mx-auto max-w-5xl space-y-6 px-4 py-6">
<PageHeader
title="数据同步"
subtitle="将本地数据推送到 serve 服务器"
/>
{/* 连接配置 */}
<section className="rounded-card border border-border bg-surface p-5">
<div className="flex items-center justify-between">
<div className="flex items-center gap-2">
{status?.config.serve_url ? (
<Cloud className="h-5 w-5 text-accent" />
) : (
<CloudOff className="h-5 w-5 text-muted" />
)}
<div>
<div className="text-sm font-medium text-foreground">Serve </div>
<div className="text-xs text-muted">
{status?.config.serve_url
? status.config.serve_url
: '未配置 — 点击右侧设置按钮'}
</div>
</div>
</div>
<div className="flex items-center gap-2">
{status?.last_sync && (
<span className="text-[10px] text-muted">
: {new Date(status.last_sync).toLocaleString('zh-CN')}
</span>
)}
<button
onClick={() => { setShowSettings(v => !v); refetch() }}
className="rounded-btn p-2 text-muted hover:text-foreground hover:bg-elevated transition-colors"
title="同步设置"
>
<Settings className="h-4 w-4" />
</button>
<button
onClick={() => refetch()}
className="rounded-btn p-2 text-muted hover:text-foreground hover:bg-elevated transition-colors"
title="刷新状态"
>
<RefreshCw className={cn('h-4 w-4', isLoading && 'animate-spin')} />
</button>
</div>
</div>
{showSettings && (
<div className="mt-4 space-y-3 border-t border-border pt-4">
<div>
<label className="text-xs text-muted">Serve </label>
<input
value={serveUrl}
onChange={e => setServeUrl(e.target.value)}
placeholder="http://your-server:3018"
className="mt-1 w-full rounded border border-border bg-base px-3 py-2 text-sm text-foreground placeholder:text-muted focus:border-accent focus:outline-none"
/>
</div>
<div>
<label className="text-xs text-muted"> Key</label>
<input
value={syncKey}
onChange={e => setSyncKey(e.target.value)}
placeholder="留空则不校验"
type="password"
className="mt-1 w-full rounded border border-border bg-base px-3 py-2 text-sm text-foreground placeholder:text-muted focus:border-accent focus:outline-none"
/>
</div>
<button
onClick={() => saveCfg.mutate()}
disabled={saveCfg.isPending}
className="inline-flex items-center gap-2 rounded-btn bg-accent px-4 py-2 text-sm font-medium text-white hover:bg-accent/90 disabled:opacity-50 transition-colors"
>
{saveCfg.isPending ? (
<Loader2 className="h-4 w-4 animate-spin" />
) : (
<Save className="h-4 w-4" />
)}
</button>
</div>
)}
</section>
{/* 数据选择 */}
<section className="rounded-card border border-border bg-surface p-5">
<div className="flex items-center justify-between">
<h3 className="text-sm font-medium text-foreground"></h3>
<div className="flex items-center gap-2">
<button
onClick={selectAll}
className="text-xs text-accent hover:text-accent/80 transition-colors"
>
</button>
<span className="text-xs text-muted">/</span>
<button
onClick={selectNone}
className="text-xs text-accent hover:text-accent/80 transition-colors"
>
</button>
</div>
</div>
<div className="mt-3 grid grid-cols-2 gap-2">
{ALL_PARTS.map(({ key, label }) => {
const info = status?.parts?.[key]
const hasData = info && info.file_count > 0
return (
<label
key={key}
className={cn(
'flex items-center gap-3 rounded-lg border px-3 py-2.5 cursor-pointer transition-colors',
selected.has(key)
? 'border-accent/40 bg-accent/5'
: 'border-border bg-base hover:border-muted',
)}
>
<input
type="checkbox"
checked={selected.has(key)}
onChange={() => togglePart(key)}
className="h-4 w-4 rounded border-border text-accent focus:ring-accent"
/>
<div className="min-w-0 flex-1">
<div className="flex items-center gap-1.5">
<span className="text-sm text-foreground">{label}</span>
{hasData && (
<HardDrive className="h-3 w-3 text-accent shrink-0" />
)}
</div>
{info && (
<div className="mt-0.5 text-[10px] text-muted">
{info.file_count > 0
? `${info.file_count} 文件 · ${fmtSize(info.size_bytes)}`
: '无数据'}
</div>
)}
</div>
</label>
)
})}
</div>
</section>
{/* 操作 */}
<section className="flex items-center justify-between rounded-card border border-border bg-surface p-5">
<div className="flex items-center gap-2 text-sm text-muted">
<Database className="h-4 w-4" />
<span>
{selected.size}
{push.data && (
<span className="ml-2 text-bull">
<CheckCircle2 className="inline h-3.5 w-3.5 mr-0.5" />
</span>
)}
</span>
</div>
<button
onClick={() => push.mutate()}
disabled={push.isPending || selected.size === 0}
className="inline-flex items-center gap-2 rounded-btn bg-accent px-5 py-2.5 text-sm font-medium text-white hover:bg-accent/90 disabled:opacity-50 transition-colors"
>
{push.isPending ? (
<>
<Loader2 className="h-4 w-4 animate-spin" />
</>
) : (
<>
<Upload className="h-4 w-4" />
</>
)}
</button>
</section>
{/* 同步日志 */}
{push.data && (
<section className="rounded-card border border-border bg-surface p-4">
<div className="text-xs text-muted font-mono leading-relaxed">
<div> </div>
<div> : {push.data.parts_sent.join(', ')}</div>
<div> : {fmtSize(push.data.size_bytes)}</div>
<div> : {push.data.serve_response.file_count}</div>
</div>
</section>
)}
{push.error && (
<section className="rounded-card border border-danger/30 bg-danger/5 p-4">
<div className="text-xs text-danger font-mono leading-relaxed">
<div> </div>
<div> {String(push.error)}</div>
</div>
</section>
)}
</div>
)
}
+2
View File
@@ -20,6 +20,7 @@ import { Branding } from './pages/Branding'
import { Settings } from './pages/Settings'
import { Indices } from './pages/Indices'
import { Dev } from './pages/Dev'
import { Sync } from './pages/Sync'
import { useSettings } from './lib/useSharedQueries'
import { Logo } from './components/Logo'
@@ -77,6 +78,7 @@ export const router = createBrowserRouter([
{ path: 'data', element: <Data /> },
{ path: 'monitor', element: <Monitor /> },
{ path: 'trading', element: <Trading /> },
{ path: 'sync', element: <Sync /> },
{ path: 'limit-ladder', element: <LimitUpLadder /> },
{ path: 'indices', element: <Indices /> },
{ path: 'branding', element: <Branding /> },
+135
View File
@@ -0,0 +1,135 @@
"""数据同步 — 接收 local 端推送的 Parquet 数据包。"""
from __future__ import annotations
import logging
import tarfile
import tempfile
from io import BytesIO
from pathlib import Path
from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, File, Form
from app.config import settings
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/data/sync", tags=["data-sync"])
# 允许同步的子目录列表
SYNCABLE_PARTS = {
"kline_daily", "kline_daily_enriched",
"kline_index_daily", "kline_index_enriched",
"kline_etf_daily", "kline_etf_enriched", "kline_etf_minute",
"kline_minute",
"adj_factor", "adj_factor_etf",
"instruments", "instruments_index", "instruments_etf",
"financials",
"pools",
"ext_data",
}
def _verify_sync_key(request: Request) -> None:
"""简单的鉴权 — 校验 X-Sync-Key 头。"""
key = request.headers.get("X-Sync-Key", "")
expected = getattr(settings, "sync_key", "")
if expected and key != expected:
raise HTTPException(status_code=403, detail="sync key mismatch")
def _refresh_views(request: Request) -> None:
"""重新注册 DuckDB 视图 + 刷新 Polars 缓存。"""
repo = getattr(request.app.state, "repo", None)
if repo is None:
return
try:
d = repo.store.data_dir.as_posix()
views = {
"kline_daily": f"{d}/kline_daily/**/*.parquet",
"kline_enriched": f"{d}/kline_daily_enriched/**/*.parquet",
"kline_index_daily": f"{d}/kline_index_daily/**/*.parquet",
"kline_index_enriched": f"{d}/kline_index_enriched/**/*.parquet",
"kline_etf_daily": f"{d}/kline_etf_daily/**/*.parquet",
"kline_etf_enriched": f"{d}/kline_etf_enriched/**/*.parquet",
"kline_etf_minute": f"{d}/kline_etf_minute/**/*.parquet",
"kline_minute": f"{d}/kline_minute/**/*.parquet",
"adj_factor": f"{d}/adj_factor/**/*.parquet",
"adj_factor_etf": f"{d}/adj_factor_etf/**/*.parquet",
"instruments": f"{d}/instruments/**/*.parquet",
"instruments_index": f"{d}/instruments_index/**/*.parquet",
"instruments_etf": f"{d}/instruments_etf/**/*.parquet",
}
for name, path in views.items():
try:
repo.store.db.execute(
f"CREATE OR REPLACE VIEW {name} AS "
f"SELECT * FROM read_parquet('{path}', union_by_name=true)"
)
except Exception as exc:
logger.warning("view %s refresh failed: %s", name, exc)
repo.store._register_unified_views()
repo.clear_cache()
repo.refresh_cache()
# 清除 API 数据缓存
from app.api.data import invalidate_data_cache
invalidate_data_cache()
except Exception as exc:
logger.warning("cache refresh failed: %s", exc)
@router.post("/upload")
async def upload_sync(
request: Request,
file: UploadFile = File(...),
parts: str = Form(""),
_auth: None = Depends(_verify_sync_key),
):
"""接收 local 端打包的 tar.gz 数据,解压写入 data/ 目录。"""
data_dir: Path = request.app.state.datastore.data_dir
# 解析要同步的 parts
requested = set(p.strip() for p in parts.split(",") if p.strip())
if not requested:
raise HTTPException(status_code=400, detail="parts 不能为空,逗号分隔子目录名")
invalid = requested - SYNCABLE_PARTS
if invalid:
raise HTTPException(
status_code=400,
detail=f"不支持的同步目录: {', '.join(sorted(invalid))}",
)
# 读取上传的 tar.gz
raw = await file.read()
if not raw:
raise HTTPException(status_code=400, detail="上传文件为空")
extracted = 0
try:
with tarfile.open(fileobj=BytesIO(raw), mode="r:gz") as tar:
for member in tar.getmembers():
# 安全校验: 防止路径穿越
member_path = Path(member.name)
if member_path.is_absolute() or ".." in member_path.parts:
logger.warning("skipping unsafe path: %s", member.name)
continue
# 确定属于哪个 part
top_dir = member_path.parts[0] if member_path.parts else ""
if top_dir not in requested:
continue
target = data_dir / member.name
if member.isfile():
target.parent.mkdir(parents=True, exist_ok=True)
with tar.extractfile(member) as src:
if src is None:
continue
target.write_bytes(src.read())
extracted += 1
except tarfile.TarError as exc:
raise HTTPException(status_code=400, detail=f"压缩包解析失败: {exc}") from exc
# 刷新缓存
_refresh_views(request)
logger.info("sync uploaded: parts=%s files=%d", parts, extracted)
return {"ok": True, "parts": sorted(requested), "file_count": extracted}
+5 -20
View File
@@ -20,11 +20,7 @@ router = APIRouter(prefix="/api/financials", tags=["financials"])
@router.get("/status")
def financial_status(request: Request):
"""返回各财务表的同步状态。无需 FINANCIAL 权限(前端根据 available 决定是否展示)。"""
capset = request.app.state.capabilities
if not capset.has(Cap.FINANCIAL):
return {"available": False, "tables": {}}
"""返回各财务表的同步状态。"""
data_dir = request.app.state.repo.store.data_dir
tables = {}
@@ -42,15 +38,16 @@ def financial_status(request: Request):
else:
tables[table] = {"rows": 0, "symbols": 0}
capset = request.app.state.capabilities
has_cap = capset.has(Cap.FINANCIAL)
fs = getattr(request.app.state, "financial_scheduler", None)
last_sync = fs.last_sync if fs else {}
has_data = any(t.get("rows", 0) > 0 for t in tables.values())
return {
"available": True,
"available": has_cap or has_data,
"tables": tables,
"last_sync": last_sync,
# 服务端是否正在同步(手动触发)——前端据此显示"同步中"并防重复点击,
# 且刷新页面后仍能正确反映服务端状态。
"syncing": bool(fs and fs.is_syncing),
}
@@ -58,9 +55,6 @@ def financial_status(request: Request):
@router.get("/metrics")
def get_metrics(request: Request, symbol: str | None = None):
"""查询核心财务指标。"""
capset = request.app.state.capabilities
capset.require(Cap.FINANCIAL)
df = get_financial_df(request.app.state.repo.store.data_dir, "metrics")
if df.is_empty():
return {"data": []}
@@ -72,9 +66,6 @@ def get_metrics(request: Request, symbol: str | None = None):
@router.get("/income")
def get_income(request: Request, symbol: str | None = None):
"""查询利润表。"""
capset = request.app.state.capabilities
capset.require(Cap.FINANCIAL)
df = get_financial_df(request.app.state.repo.store.data_dir, "income")
if df.is_empty():
return {"data": []}
@@ -86,9 +77,6 @@ def get_income(request: Request, symbol: str | None = None):
@router.get("/balance-sheet")
def get_balance_sheet(request: Request, symbol: str | None = None):
"""查询资产负债表。"""
capset = request.app.state.capabilities
capset.require(Cap.FINANCIAL)
df = get_financial_df(request.app.state.repo.store.data_dir, "balance_sheet")
if df.is_empty():
return {"data": []}
@@ -100,9 +88,6 @@ def get_balance_sheet(request: Request, symbol: str | None = None):
@router.get("/cash-flow")
def get_cash_flow(request: Request, symbol: str | None = None):
"""查询现金流量表。"""
capset = request.app.state.capabilities
capset.require(Cap.FINANCIAL)
df = get_financial_df(request.app.state.repo.store.data_dir, "cash_flow")
if df.is_empty():
return {"data": []}
+3
View File
@@ -97,6 +97,9 @@ class Settings(BaseSettings):
# 公网服务器部署时免去 SSH 端口转发设密码的麻烦。写入 auth.json(哈希)后即不再读取。
auth_password: str = ""
# Sync — 数据同步鉴权 Keylocal→serve 推送时校验 X-Sync-Key 头)
sync_key: str = ""
# Data — frozen: exe 同级 data/ 子目录; 非 frozen: 项目根 data/
# (均可被环境变量 DATA_DIR 覆盖, pydantic-settings 自动注入)
data_dir: Path = _user_data_root()
+3 -2
View File
@@ -11,7 +11,7 @@ from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from app import __version__
from app.api import analysis, auth as auth_api, data, ext_data, financials, indices, kline, market_recap, alerts, overview, pipeline, rps, screener, settings as settings_api, signals, stock_analysis, strategy
from app.api import analysis, auth as auth_api, data, data_sync, ext_data, financials, indices, kline, market_recap, alerts, overview, pipeline, rps, screener, settings as settings_api, signals, stock_analysis, strategy
from app.api.routes import router as core_router
from app.config import settings
from app.jobs import daily_pipeline
@@ -141,7 +141,7 @@ app.add_middleware(
# 2. 未设密码 + 公网 → 拒绝(403, 防裸奔也防抢占; 引导本机设密码)
# 3. 已设密码 → 检查 session, 无效则 401(前端跳登录)
# 白名单: /api/auth/* (设密码/登录本身)、/health 等探活。
_AUTH_WHITELIST_PREFIX = ("/api/auth/",)
_AUTH_WHITELIST_PREFIX = ("/api/auth/", "/api/data/sync/")
_AUTH_WHITELIST_EXACT = ("/health", "/api/health", "/openapi.json", "/docs", "/redoc")
@@ -188,6 +188,7 @@ app.include_router(overview.router)
app.include_router(analysis.router)
app.include_router(pipeline.router)
app.include_router(data.router)
app.include_router(data_sync.router)
app.include_router(ext_data.router)
app.include_router(financials.router)
app.include_router(stock_analysis.router)
+21 -1
View File
@@ -57,7 +57,20 @@ export function Financials() {
rememberStock(symbol, name)
}
// 无 Expert 能力时: 等 status 加载完, 判断是否有从 local 同步来的数据
if (!hasFinancial) {
if (isLoading) {
return (
<>
<PageHeader title="财务分析" subtitle="利润表 / 资负表 / 现金流 / 关键指标 / AI分析 · Expert" />
<div className="flex items-center justify-center py-16">
<Loader2 className="h-5 w-5 animate-spin text-muted" />
</div>
</>
)
}
if (!status?.available) {
// 既无 Expert 能力, 也无同步数据 → 锁定页
return (
<>
<PageHeader title="财务分析" subtitle="利润表 / 资负表 / 现金流 / 关键指标 / AI分析 · Expert" />
@@ -70,7 +83,6 @@ export function Financials() {
<p className="mt-2 text-xs leading-relaxed text-secondary">
Expert
</p>
{/* 当前财务数据源(TickFlow)需付费,后续将接入免费数据源;期间欢迎在 issues 推荐免费源 */}
<div className="mt-5 rounded-btn border border-accent/25 bg-accent/[0.05] px-3.5 py-3 text-left">
<div className="flex items-center gap-1.5 text-xs font-medium text-accent">
<Lightbulb className="h-3.5 w-3.5 shrink-0" />
@@ -94,6 +106,8 @@ export function Financials() {
</>
)
}
// 有同步数据, 继续展示
}
const handleSync = (table: string) => {
// 防重复点击:syncing 中不再触发(后端 trigger 也有 _is_syncing 兜底)
@@ -171,6 +185,7 @@ export function Financials() {
: '同步中…'}
</span>
)}
{hasFinancial && (
<button
className="inline-flex items-center gap-1.5 px-3 py-1.5 rounded-btn bg-gradient-to-r from-accent/25 to-accent/10 border border-accent/30 text-accent text-xs font-medium hover:from-accent/35 hover:to-accent/20 transition-all duration-150 disabled:opacity-40 disabled:cursor-not-allowed"
onClick={() => handleSync('all')}
@@ -182,6 +197,7 @@ export function Financials() {
: <RefreshCw className="h-3.5 w-3.5" />}
{syncing ? '同步中…' : '全部同步'}
</button>
)}
</div>
}
/>
@@ -233,6 +249,7 @@ export function Financials() {
)}
<span className="text-xs font-medium text-foreground">{label}</span>
</div>
{hasFinancial ? (
<button
className="text-muted hover:text-accent transition-colors disabled:opacity-30 disabled:cursor-not-allowed"
onClick={() => handleSync(key)}
@@ -243,6 +260,9 @@ export function Financials() {
? <Loader2 className="h-3.5 w-3.5 animate-spin" />
: <RefreshCw className="h-3.5 w-3.5" />}
</button>
) : (
<div className="h-3.5 w-3.5" />
)}
</div>
<div className="mt-2 text-xl font-semibold tabular-nums text-foreground">
{fmtBigNum(info?.rows ?? 0)}