Initial commit: 可转债轮动分析系统

- 实时行情:东财push2主源 + 腾讯批量报价备用源(自动切换)
- 历史K线:东财push2his主源 + 腾讯web子域备用源(含熔断机制)
- 增量同步:支持 -incr/-since 参数按存档日期增量入库
- 策略回测:净值曲线/10指标/基准对比,含基本面过滤
- 用户认证:用户名密码 + 邮箱验证码双登录方式
- 前端:三Tab SPA(实时行情/策略回测/数据管理)
This commit is contained in:
2026-08-17 20:23:22 +08:00
commit e28bfec957
45 changed files with 7889 additions and 0 deletions

258
cmd/backtest/main.go Normal file
View File

@@ -0,0 +1,258 @@
// backtest 是可转债轮动策略回测命令。
//
// go run ./cmd/backtest [-db 库路径] [-from 2020-01-01] [-to 2025-12-31] \
// [-strategy lowprice|doublelow] [-topn 10] [-rebalance 21] \
// [-min-price 80] [-max-price 130] [-max-premium 30] [-cash 100000] [-out 曲线.csv]
//
// 从 SQLite 加载全量转债日K线,运行内置轮动策略(低价/双低),
// 输出绩效指标并与等权全转债基准对比。
package main
import (
"context"
"encoding/csv"
"flag"
"fmt"
"os"
"path/filepath"
"stock-analysis/internal/backtest"
"stock-analysis/internal/strategy"
"stock-analysis/internal/storage/sqlite"
)
func main() {
dbPath := flag.String("db", defaultDBPath(), "SQLite 数据库文件路径")
from := flag.String("from", "2020-01-01", "回测起始日(含)")
to := flag.String("to", "", "回测结束日(含,默认数据末日)")
topn := flag.Int("topn", 10, "持仓数量")
rebalance := flag.Int("rebalance", 21, "调仓周期(交易日)")
minPrice := flag.Float64("min-price", 80, "价格下限")
maxPrice := flag.Float64("max-price", 130, "价格上限")
strategyName := flag.String("strategy", "lowprice", "策略: lowprice(低价轮动)|doublelow(双低轮动)|premium(溢价率轮动)")
maxPremium := flag.Float64("max-premium", 30, "双低策略溢价率上限")
fundamental := flag.Bool("fundamental", false, "启用正股基本面过滤器(排除ST/高负债/亏损标的)")
cash := flag.Float64("cash", 100000, "初始资金(元)")
out := flag.String("out", "", "净值曲线 CSV 输出路径(可选)")
flag.Parse()
ctx := context.Background()
store, err := sqlite.Open(*dbPath)
if err != nil {
fatal("打开数据库失败: ", err)
}
defer store.Close()
feed, err := backtest.NewDataFeed(ctx, store)
if err != nil {
fatal("加载K线数据失败: ", err)
}
fmt.Printf("数据加载完成:%d 个交易日(%s ~ %s)\n\n",
len(feed.Dates()), feed.Dates()[0], feed.Dates()[len(feed.Dates())-1])
params := backtest.DefaultParams()
params.InitialCash = *cash
params.StartDate = *from
params.EndDate = *to
var strat backtest.Strategy
switch *strategyName {
case "lowprice":
strat = strategy.NewLowPrice(strategy.LowPriceParams{
TopN: *topn,
Rebalance: *rebalance,
MinPrice: *minPrice,
MaxPrice: *maxPrice,
})
case "doublelow":
strat = strategy.NewDoubleLow(strategy.DoubleLowParams{
TopN: *topn,
Rebalance: *rebalance,
MinPrice: *minPrice,
MaxPrice: *maxPrice,
MaxPremium: *maxPremium,
})
case "premium":
strat = strategy.NewPremium(strategy.PremiumParams{
TopN: *topn,
Rebalance: *rebalance,
MinPrice: *minPrice,
MaxPrice: *maxPrice,
})
default:
fatal("未知策略 ", fmt.Errorf("应为 lowprice、doublelow 或 premium"))
}
engine := backtest.NewEngine(params, strat)
if *fundamental {
// 加载正股基本面数据
fundamentals, err := store.LoadAllFundamentals(ctx)
if err != nil {
fatal("加载基本面数据失败: ", err)
}
// 获取正股-转债映射
stock2bond, err := store.GetStockBondMap(ctx)
if err != nil {
fatal("获取正股-转债映射失败: ", err)
}
filter := backtest.NewFundamentalFilter(stock2bond)
engine.SetFundamentalFilter(filter, fundamentals)
fmt.Printf("基本面过滤器已启用:%d 只正股数据, %d 条正股-转债映射\n", len(fundamentals), len(stock2bond))
}
res, err := engine.Run(ctx, feed)
if err != nil {
fatal("回测失败: ", err)
}
fmt.Printf("策略:%s(%s ~ %s)\n", res.Strategy, res.Start, res.End)
printMetrics("策略", res.Metrics)
bench := computeBenchmark(feed, *from, *to)
fmt.Println()
fmt.Printf("基准:等权全转债指数(%s ~ %s)\n", *from, *to)
printMetrics("基准", bench)
if *out != "" {
if err := writeCSV(*out, res, feed); err != nil {
fatal("写CSV失败: ", err)
}
fmt.Printf("\n净值曲线已输出到 %s\n", *out)
}
}
// computeBenchmark 计算等权全转债指数:每日对全部存续转债取等权平均日收益并复利。
func computeBenchmark(feed *backtest.DataFeed, from, to string) backtest.Metrics {
var (
idx float64 = 1
prev map[string]float64
curve []backtest.EquityPoint
hasPrev bool
)
for _, date := range feed.Dates() {
if from != "" && date < from {
continue
}
if to != "" && date > to {
break
}
cur := make(map[string]float64)
for _, q := range feed.QuotesOn(date) {
cur[q.Code] = q.Price
}
if hasPrev {
var sum, n float64
for code, p := range cur {
if pp, ok := prev[code]; ok && pp > 0 {
sum += p/pp - 1
n++
}
}
if n > 0 {
idx *= 1 + sum/n
}
}
curve = append(curve, backtest.EquityPoint{Date: date, Value: idx, Return: idx - 1})
prev = cur
hasPrev = true
}
return backtest.ComputeMetrics(curve)
}
// printMetrics 打印绩效指标表格。
func printMetrics(label string, m backtest.Metrics) {
fmt.Printf("%-14s 总收益 %9.2f%% 年化 %8.2f%% 最大回撤 %7.2f%% 夏普 %6.2f 波动率 %5.2f%% 胜率 %5.1f%%\n",
label,
m.TotalReturn*100, m.AnnualReturn*100, m.MaxDrawdown*100,
m.Sharpe, m.Volatility*100, m.WinRatio*100)
}
// writeCSV 输出策略与基准的净值曲线(含累计收益)到 CSV 文件。
func writeCSV(path string, res *backtest.Result, feed *backtest.DataFeed) error {
f, err := os.Create(path)
if err != nil {
return err
}
defer f.Close()
w := csv.NewWriter(f)
defer w.Flush()
if err := w.Write([]string{"date", "strategy_value", "strategy_return", "bench_value", "bench_return"}); err != nil {
return err
}
// 基准曲线独立重算,与策略日期对齐。
benchCurve := benchmarkCurve(feed, res.Start, res.End)
byDate := make(map[string]backtest.EquityPoint, len(benchCurve))
for _, p := range benchCurve {
byDate[p.Date] = p
}
for _, p := range res.Curve {
b, ok := byDate[p.Date]
bv, br := 1.0, 0.0
if ok {
bv, br = b.Value, b.Return
}
if err := w.Write([]string{
p.Date,
fmt.Sprintf("%.2f", p.Value),
fmt.Sprintf("%.6f", p.Return),
fmt.Sprintf("%.6f", bv),
fmt.Sprintf("%.6f", br),
}); err != nil {
return err
}
}
return nil
}
// benchmarkCurve 返回 from~to 区间的基准净值曲线。
func benchmarkCurve(feed *backtest.DataFeed, from, to string) []backtest.EquityPoint {
var (
idx float64 = 1
prev map[string]float64
curve []backtest.EquityPoint
hasPrev bool
)
for _, date := range feed.Dates() {
if date < from || (to != "" && date > to) {
continue
}
cur := make(map[string]float64)
for _, q := range feed.QuotesOn(date) {
cur[q.Code] = q.Price
}
if hasPrev {
var sum, n float64
for code, p := range cur {
if pp, ok := prev[code]; ok && pp > 0 {
sum += p/pp - 1
n++
}
}
if n > 0 {
idx *= 1 + sum/n
}
}
curve = append(curve, backtest.EquityPoint{Date: date, Value: idx, Return: idx - 1})
prev = cur
hasPrev = true
}
return curve
}
// defaultDBPath 与 ingest 保持一致,返回默认数据库路径。
func defaultDBPath() string {
local := os.Getenv("LOCALAPPDATA")
if local == "" {
local = os.TempDir()
}
return filepath.Join(local, "stock-analysis", "stock.db")
}
// fatal 打印错误并退出。
func fatal(msg string, err error) {
fmt.Fprintln(os.Stderr, msg, err)
os.Exit(1)
}

41
cmd/dbstats/main.go Normal file
View File

@@ -0,0 +1,41 @@
// dbstats 输出 SQLite 库的转债数与K线总数统计。
//
// go run ./cmd/dbstats [-db 库路径]
package main
import (
"context"
"database/sql"
"flag"
"fmt"
"os"
_ "modernc.org/sqlite"
)
func main() {
dbPath := flag.String("db", "C:\\Users\\zhouww\\AppData\\Local\\Temp\\stock-analysis-data\\stock.db", "SQLite 数据库文件路径")
flag.Parse()
db, err := sql.Open("sqlite", *dbPath)
if err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
defer db.Close()
db.SetMaxOpenConns(1)
ctx := context.Background()
var bonds, rows int
if err := db.QueryRowContext(ctx, "SELECT COUNT(DISTINCT code), COUNT(*) FROM convertible_kline").Scan(&bonds, &rows); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
fmt.Printf("转债数: %d,K线总数: %d(库: %s)\n", bonds, rows, *dbPath)
// 数据范围抽查。
var minDate, maxDate string
if err := db.QueryRowContext(ctx, "SELECT MIN(date), MAX(date) FROM convertible_kline").Scan(&minDate, &maxDate); err == nil {
fmt.Printf("K线日期范围: %s ~ %s\n", minDate, maxDate)
}
}

85
cmd/fetch/main.go Normal file
View File

@@ -0,0 +1,85 @@
// fetch 是数据拉取验证命令(CLI):
//
// go run ./cmd/fetch # 拉取全量可转债实时行情
// go run ./cmd/fetch -code 113667 # 拉取指定可转债历史K线
package main
import (
"context"
"flag"
"fmt"
"os"
"sort"
"stock-analysis/internal/datasource/eastmoney"
"stock-analysis/internal/datasource/fallback"
"stock-analysis/internal/datasource/tencent"
"stock-analysis/internal/service"
)
func main() {
code := flag.String("code", "", "可转债代码(如 113667),指定时拉取历史K线")
days := flag.Int("days", 30, "打印最近N根K线")
flag.Parse()
ds := fallback.New(eastmoney.NewClient(), tencent.NewClient())
svc := service.NewMarketService(ds)
ctx := context.Background()
if *code != "" {
if err := fetchKline(ctx, svc, *code, *days); err != nil {
fmt.Fprintln(os.Stderr, "kline error:", err)
os.Exit(1)
}
return
}
if err := fetchConvertibles(ctx, svc); err != nil {
fmt.Fprintln(os.Stderr, "list error:", err)
os.Exit(1)
}
}
// fetchConvertibles 拉取并打印全量可转债。
func fetchConvertibles(ctx context.Context, svc *service.MarketService) error {
bonds, err := svc.ListConvertibles(ctx)
if err != nil {
return err
}
fmt.Printf("共拉取可转债 %d 只\n\n", len(bonds))
// 按溢价率升序展示前 15 只(低溢价候选池雏形)。
sort.Slice(bonds, func(i, j int) bool { return bonds[i].PremiumRate < bonds[j].PremiumRate })
fmt.Printf("%-8s %-10s %8s %7s %8s %8s %8s %8s %8s\n",
"代码", "名称", "价格", "涨跌%", "溢价率%", "转股价值", "转股价", "纯债价值", "正股")
for i, b := range bonds {
if i >= 15 {
break
}
fmt.Printf("%-8s %-10s %8.3f %7.2f %8.2f %8.2f %8.2f %8.2f %s\n",
b.Code, b.Name, b.Price, b.ChangePct, b.PremiumRate,
b.ConvertValue, b.ConvertPrice, b.PureBondVal, b.StockName)
}
fmt.Println("\n... (仅展示前15只,按溢价率升序)")
return nil
}
// fetchKline 拉取并打印指定可转债的历史K线。
func fetchKline(ctx context.Context, svc *service.MarketService, code string, days int) error {
klines, err := svc.GetDailyKline(ctx, code, "20200101")
if err != nil {
return err
}
fmt.Printf("代码 %s 共 %d 根日K(%s ~ %s)\n\n",
code, len(klines), klines[0].Date, klines[len(klines)-1].Date)
start := len(klines) - days
if start < 0 {
start = 0
}
fmt.Printf("%-12s %8s %8s %8s %8s %12s\n", "日期", "开盘", "收盘", "最高", "最低", "成交量")
for _, k := range klines[start:] {
fmt.Printf("%-12s %8.3f %8.3f %8.3f %8.3f %12d\n",
k.Date, k.Open, k.Close, k.High, k.Low, k.Volume)
}
return nil
}

102
cmd/fundamentals/main.go Normal file
View File

@@ -0,0 +1,102 @@
// fundamentals 是正股基本面数据入库命令:
//
// go run ./cmd/fundamentals [-db 库路径]
//
// 从 convert_bond 表读取所有正股代码,通过东财 clist 接口批量拉取
// 基本面数据(ROE/资产负债率/净利润/ST状态),写入 stock_fundamental 表。
package main
import (
"context"
"flag"
"fmt"
"os"
"path/filepath"
"time"
"stock-analysis/internal/datasource/eastmoney"
"stock-analysis/internal/storage/sqlite"
)
func main() {
dbPath := flag.String("db", defaultDBPath(), "SQLite 数据库文件路径")
flag.Parse()
store, err := sqlite.Open(*dbPath)
if err != nil {
fatal("打开数据库失败: ", err)
}
defer store.Close()
ctx := context.Background()
// 1. 获取所有正股代码。
stock2bond, err := store.GetStockBondMap(ctx)
if err != nil {
fatal("获取正股映射失败: ", err)
}
if len(stock2bond) == 0 {
fmt.Println("convert_bond 表为空,请先执行 go run ./cmd/ingest -all 入库转债数据")
return
}
stockCodes := make([]string, 0, len(stock2bond))
for code := range stock2bond {
stockCodes = append(stockCodes, code)
}
fmt.Printf("正股数量:%d\n", len(stockCodes))
// 2. 分批拉取基本面数据(每批最多 200 只)。
client := eastmoney.NewClient()
const batchSize = 200
var totalSaved, totalFail int
for i := 0; i < len(stockCodes); i += batchSize {
end := i + batchSize
if end > len(stockCodes) {
end = len(stockCodes)
}
batch := stockCodes[i:end]
funds, err := client.FetchStockFundamentals(ctx, batch)
if err != nil {
fmt.Printf("批次 %d-%d 拉取失败: %v\n", i, end-1, err)
totalFail += len(batch)
continue
}
for _, f := range funds {
f.UpdateDate = time.Now().Format("2006-01-02")
if err := store.SaveStockFundamental(ctx, f); err != nil {
fmt.Printf(" %s %s: 保存失败 %v\n", f.Code, f.Name, err)
totalFail++
continue
}
totalSaved++
}
fmt.Printf("批次 %d-%d: 拉取 %d 只, 已保存 %d\n", i+1, end, len(batch), totalSaved)
// 控制请求频率。
if end < len(stockCodes) {
time.Sleep(2 * time.Second)
}
}
fmt.Printf("\n基本面入库完成:成功 %d 只,失败 %d 只(库: %s)\n", totalSaved, totalFail, *dbPath)
}
// defaultDBPath 返回默认数据库路径。
func defaultDBPath() string {
local := os.Getenv("LOCALAPPDATA")
if local == "" {
local = os.TempDir()
}
return filepath.Join(local, "stock-analysis", "stock.db")
}
// fatal 打印错误并退出。
func fatal(msg string, err error) {
fmt.Fprintln(os.Stderr, msg, err)
os.Exit(1)
}

369
cmd/ingest/main.go Normal file
View File

@@ -0,0 +1,369 @@
// ingest 是可转债历史K线入库命令:
//
// go run ./cmd/ingest -code 110062 -from 2020 [-source eastmoney|tencent]
// go run ./cmd/ingest -latest 10 [-kline-source tencent]
// go run ./cmd/ingest -all [-source eastmoney] // 全市场转债历史数据入库
// go run ./cmd/ingest -incr // 增量同步:从库中最后日期起拉取最新数据
// go run ./cmd/ingest -incr -since 20260101 // 增量同步:从指定日期起拉取
// go run ./cmd/ingest -incr -all -limit 10 // 增量同步:仅处理前 10 只已入库转债
//
// 分批拉取可转债全量日K线,幂等写入 SQLite(默认 %LOCALAPPDATA%\stock-analysis\stock.db)。
// 默认使用东财行情源;东财 push2 接口被限流时可切换 tencent 备用源。
// 客户端内置 2s 请求间隔 + 多域名轮询 + 重试,遵循公开接口限速要求。
package main
import (
"context"
"flag"
"fmt"
"os"
"path/filepath"
"strconv"
"time"
"stock-analysis/internal/datasource"
"stock-analysis/internal/datasource/eastmoney"
"stock-analysis/internal/datasource/fallback"
"stock-analysis/internal/datasource/tencent"
"stock-analysis/internal/model"
"stock-analysis/internal/service"
"stock-analysis/internal/storage/sqlite"
)
func main() {
code := flag.String("code", "110062", "可转债代码(默认烽火转债)")
fromYear := flag.Int("from", 2020, "起始年份(默认 2020)")
latest := flag.Int("latest", 0, "批量拉取最新上市的前 N 只可转债入库(设置后忽略 -code/-from)")
all := flag.Bool("all", false, "拉取全市场可转债历史数据入库(设置后忽略其它参数)")
limit := flag.Int("limit", 0, "-all / -incr -all 模式下最多处理的转债数量(0 表示不限,用于小范围测试)")
incr := flag.Bool("incr", false, "增量同步模式:从库中最后日期(或 -since 指定日期)起拉取最新K线入库")
since := flag.String("since", "", "增量同步起点(YYYYMMDD,如 20260101);不指定则自动取库中最后日期的次日")
dbPath := flag.String("db", defaultDBPath(), "SQLite 数据库文件路径")
source := flag.String("source", "eastmoney", "K线数据源:eastmoney(东财)/ tencent(腾讯备用)")
flag.Parse()
var err error
switch {
case *incr:
err = runIncremental(*code, *all, *since, *limit, *dbPath, *source)
case *all:
err = runAll(*dbPath, *source, *limit)
case *latest > 0:
err = runLatest(*latest, *dbPath, *source)
default:
err = run(*code, *fromYear, *dbPath, *source)
}
if err != nil {
fmt.Fprintln(os.Stderr, "ingest error:", err)
os.Exit(1)
}
}
// run 执行单只转债的分批拉取并入库。
func run(code string, fromYear int, dbPath, source string) error {
store, err := openStore(dbPath)
if err != nil {
return err
}
defer store.Close()
svc := service.NewMarketService(newDatasource(source))
ctx := context.Background()
fmt.Printf("[%s] 分批拉取 %s 历史日K(%d 年至今)...\n", source, code, fromYear)
klines, err := svc.FetchAllDailyKline(ctx, code, fromYear)
if err != nil {
return err
}
if len(klines) == 0 {
return fmt.Errorf("未获取到 %s 的历史数据", code)
}
fmt.Printf("拉取完成:共 %d 根(%s ~ %s)\n", len(klines), klines[0].Date, klines[len(klines)-1].Date)
if err := store.SaveKlines(ctx, code, klines); err != nil {
return err
}
return printCount(ctx, store, code)
}
// runLatest 批量拉取最新上市的前 n 只可转债并入库。
// 列表来源:东财 datacenter;K线来源:-source 指定(默认 eastmoney,可用 tencent)。
func runLatest(n int, dbPath, source string) error {
store, err := openStore(dbPath)
if err != nil {
return err
}
defer store.Close()
ctx := context.Background()
bonds, err := eastmoney.NewClient().FetchLatestBonds(ctx, n)
if err != nil {
return fmt.Errorf("获取最新转债列表: %w", err)
}
svc := service.NewMarketService(newDatasource(source))
var okCount, failCount int
for _, b := range bonds {
fromYear := yearOf(b.ListDate)
klines, err := svc.FetchAllDailyKline(ctx, b.Code, fromYear)
if err != nil || len(klines) == 0 {
failCount++
fmt.Printf(" %s %s 上市 %s: 无数据(%v)\n", b.Code, b.Name, b.ListDate, err)
continue
}
if err := store.SaveKlines(ctx, b.Code, klines); err != nil {
failCount++
fmt.Printf(" %s %s: 入库失败 %v\n", b.Code, b.Name, err)
continue
}
okCount++
fmt.Printf(" %s %s 上市 %s: %d 根(%s ~ %s)\n", b.Code, b.Name, b.ListDate,
len(klines), klines[0].Date, klines[len(klines)-1].Date)
}
fmt.Printf("批量入库完成:成功 %d 只,失败 %d 只(%s)\n", okCount, failCount, dbPath)
return nil
}
// runAll 拉取全市场可转债的历史数据入库。
// 流程:全量列表 → 过滤未上市/停牌 → 逐只单次请求全历史K线 → 幂等写入。
// 已入库的转债自动跳过(断点续跑),失败项会汇总列出,可重跑补齐。
func runAll(dbPath, source string, limit int) error {
store, err := openStore(dbPath)
if err != nil {
return err
}
defer store.Close()
ctx := context.Background()
ds := newDatasource(source)
svc := service.NewMarketService(ds)
fmt.Println("拉取全量转债代码列表(datacenter)...")
bonds, err := eastmoney.NewClient().FetchAllBondCodes(ctx)
if err != nil {
return fmt.Errorf("获取转债列表: %w", err)
}
// datacenter 列表包含已退市债(K线接口对其返回空,拉取时跳过)。
var candidates []model.ConvertibleBond
for _, b := range bonds {
if b.ListDate == "" {
continue
}
candidates = append(candidates, b)
}
fmt.Printf("列表共 %d 只(含退市,无发行日期已过滤 %d 只)\n", len(candidates), len(bonds)-len(candidates))
var okCount, skipCount, failCount int
var failed []string
for i, b := range candidates {
if limit > 0 && i >= limit {
fmt.Printf("达到 -limit %d 限制,提前结束\n", limit)
break
}
// 断点续跑:已入库的跳过。
if n, err := store.CountKlines(ctx, b.Code); err == nil && n > 0 {
skipCount++
fmt.Printf("[%3d/%d] %s %s 已入库(%d 根),跳过\n", i+1, len(candidates), b.Code, b.Name, n)
continue
}
klines, err := svc.FetchFullKline(ctx, b.Code, yearOf(b.ListDate))
if err != nil || len(klines) == 0 {
failCount++
failed = append(failed, b.Code)
fmt.Printf("[%3d/%d] %s %s: 无数据(%v)\n", i+1, len(candidates), b.Code, b.Name, err)
continue
}
if err := store.SaveKlines(ctx, b.Code, klines); err != nil {
failCount++
failed = append(failed, b.Code)
fmt.Printf("[%3d/%d] %s %s: 入库失败 %v\n", i+1, len(candidates), b.Code, b.Name, err)
continue
}
okCount++
fmt.Printf("[%3d/%d] %s %s: %d 根(%s ~ %s)\n", i+1, len(candidates), b.Code, b.Name,
len(klines), klines[0].Date, klines[len(klines)-1].Date)
}
fmt.Printf("全量入库完成:成功 %d,跳过 %d,失败 %d(库: %s)\n", okCount, skipCount, failCount, dbPath)
if len(failed) > 0 {
fmt.Println("失败清单(可重跑续传):", failed)
}
return nil
}
// openStore 打开 SQLite 存储,确保目录存在。
func openStore(dbPath string) (*sqlite.Store, error) {
if err := os.MkdirAll(filepath.Dir(dbPath), 0o755); err != nil {
return nil, err
}
return sqlite.Open(dbPath)
}
// runIncremental 增量同步:从库中最后日期(或 -since 指定日期)起,拉取最新K线入库。
//
// 模式:
// - 单只:-incr -code 113667 [-since 20260101]
// - 全量:-incr -all [-limit N] [-since 20260101]:对库中所有已入库转债做增量
//
// 起点确定(优先级):
// 1. -since 指定日期,作为该日期起的覆盖式拉取(含当日)
// 2. 自动取该转债在库中最后日期的次日,避免重复请求
//
// 配合 SaveKlines 的幂等语义(ON CONFLICT DO NOTHING),即使 -since 与已存档区间重叠也不会破坏数据。
func runIncremental(code string, all bool, since string, limit int, dbPath, source string) error {
store, err := openStore(dbPath)
if err != nil {
return err
}
defer store.Close()
ctx := context.Background()
svc := service.NewMarketService(newDatasource(source))
if !all {
// 单只增量同步
n, err := incrementalOne(ctx, store, svc, code, since)
if err != nil {
return err
}
if n < 0 {
fmt.Printf("%s 库中无存档数据,跳过(请先用 -all 全量入库)\n", code)
} else if n == 0 {
fmt.Printf("%s 已是最新,无需更新\n", code)
} else {
fmt.Printf("%s 增量入库 %d 根\n", code, n)
}
return nil
}
// 全量增量:遍历库中已存档的所有转债代码
codes, err := store.ListCodes(ctx)
if err != nil {
return fmt.Errorf("加载库中转债列表: %w", err)
}
fmt.Printf("增量同步开始:共 %d 只已入库转债\n", len(codes))
var okCount, skipCount, failCount int
var failed []string
for i, c := range codes {
if limit > 0 && i >= limit {
fmt.Printf("达到 -limit %d 限制,提前结束\n", limit)
break
}
n, err := incrementalOne(ctx, store, svc, c, since)
switch {
case err != nil:
failCount++
failed = append(failed, c)
fmt.Printf("[%3d/%d] %s: 失败 %v\n", i+1, len(codes), c, err)
case n < 0:
skipCount++ // 无存档日期或起点过晚,已跳过
default:
okCount++
if n > 0 {
fmt.Printf("[%3d/%d] %s: 增量入库 %d 根\n", i+1, len(codes), c, n)
} else {
skipCount++
}
}
}
fmt.Printf("增量同步完成:成功 %d,跳过 %d,失败 %d(库: %s)\n", okCount, skipCount, failCount, dbPath)
if len(failed) > 0 {
fmt.Println("失败清单(可重跑续传):", failed)
}
return nil
}
// incrementalOne 对单只转债做增量同步。
// 返回:拉取到的K线条数(>=0);-1 表示跳过(库中无存档日期)。
// 注:由于 SaveKlines 幂等语义(ON CONFLICT DO NOTHING),返回的条数可能包含已存档的日期;
// 但因 beg 已是「最后日期的次日」,正常情况下返回值即新增条数。
func incrementalOne(ctx context.Context, store *sqlite.Store, svc *service.MarketService, code, since string) (int, error) {
beg, err := resolveIncrementalBeg(ctx, store, code, since)
if err != nil {
return 0, err
}
if beg == "" {
return -1, nil
}
klines, err := svc.FetchIncrementalKline(ctx, code, beg)
if err != nil {
return 0, err
}
if len(klines) == 0 {
return 0, nil
}
if err := store.SaveKlines(ctx, code, klines); err != nil {
return 0, err
}
return len(klines), nil
}
// resolveIncrementalBeg 确定增量拉取的起始日期(YYYYMMDD)。
//
// -since 指定 → 直接使用(覆盖库中最后日期)
// 未指定 → 取库中最后日期的次日;无记录返回空串(表示该转债无存档,应跳过)
func resolveIncrementalBeg(ctx context.Context, store *sqlite.Store, code, since string) (string, error) {
if since != "" {
return since, nil
}
lastDate, err := store.LastKlineDate(ctx, code)
if err != nil {
return "", err
}
if lastDate == "" {
return "", nil
}
// lastDate 格式 YYYY-MM-DD,转 YYYYMMDD 并 +1 天,避免重复请求当天数据
t, err := time.Parse("2006-01-02", lastDate)
if err != nil {
return "", fmt.Errorf("解析最后日期 %s: %w", lastDate, err)
}
return t.AddDate(0, 0, 1).Format("20060102"), nil
}
// defaultDBPath 返回默认数据库路径。
// 项目位于 OneDrive 同步目录,SQLite 写入会受 OneDrive 文件操作限制,
// 故默认存放到本地应用数据目录(%LOCALAPPDATA%\stock-analysis\stock.db)。
func defaultDBPath() string {
local := os.Getenv("LOCALAPPDATA")
if local == "" {
local = os.TempDir()
}
return filepath.Join(local, "stock-analysis", "stock.db")
}
// printCount 打印库中指定代码的K线总数。
func printCount(ctx context.Context, store *sqlite.Store, code string) error {
total, err := store.CountKlines(ctx, code)
if err != nil {
return err
}
fmt.Printf("入库成功:库中 %s 共 %d 根日K\n", code, total)
return nil
}
// yearOf 从上市日期(2006-01-02)解析年份。
func yearOf(listDate string) int {
if len(listDate) >= 4 {
if y, err := strconv.Atoi(listDate[:4]); err == nil {
return y
}
}
return 2026
}
// newDatasource 根据名称创建K线数据源。
// 默认使用东财为主数据源,腾讯为备用数据源(东财失败时自动切换)。
func newDatasource(name string) datasource.MarketDataSource {
switch name {
case "tencent":
return tencent.NewClient()
case "eastmoney":
return fallback.New(eastmoney.NewClient(), tencent.NewClient())
default:
return fallback.New(eastmoney.NewClient(), tencent.NewClient())
}
}

87
cmd/server/main.go Normal file
View File

@@ -0,0 +1,87 @@
// server 是可转债轮动分析系统的 B/S 服务入口。
package main
import (
"context"
"errors"
"log"
"net/http"
"os"
"os/signal"
"path/filepath"
"syscall"
"time"
"stock-analysis/internal/controller"
"stock-analysis/internal/datasource/eastmoney"
"stock-analysis/internal/datasource/fallback"
"stock-analysis/internal/datasource/tencent"
"stock-analysis/internal/router"
"stock-analysis/internal/service"
"stock-analysis/internal/storage/sqlite"
)
func main() {
addr := ":8080"
if p := os.Getenv("PORT"); p != "" {
addr = ":" + p
}
webDir := "web"
dbPath := defaultDBPath()
// 东财为主数据源(实时行情+历史K线),腾讯为历史K线备用源(东财失败时自动切换)
ds := fallback.New(eastmoney.NewClient(), tencent.NewClient())
svc := service.NewMarketService(ds)
mc := controller.NewMarketController(svc, dbPath)
store, err := sqlite.Open(dbPath)
if err != nil {
log.Fatalf("open sqlite: %v", err)
}
defer store.Close()
authSvc := service.NewAuthService(store)
if err := authSvc.EnsureAdmin(context.Background()); err != nil {
log.Printf("ensure admin: %v", err)
}
ac := controller.NewAuthController(authSvc)
handler := router.New(mc, ac, webDir)
server := &http.Server{
Addr: addr,
Handler: handler,
ReadHeaderTimeout: 5 * time.Second,
}
// 优雅退出:收到信号后关闭服务。
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
go func() {
log.Printf("stock-analysis server listening on http://localhost%s", addr)
if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Fatalf("server: %v", err)
}
}()
<-ctx.Done()
log.Println("shutting down server...")
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := server.Shutdown(shutdownCtx); err != nil {
log.Printf("shutdown: %v", err)
}
}
// defaultDBPath 返回默认数据库路径。
// 默认使用项目目录下的 stock.db,避免 OneDrive/AppData 的 I/O 限制。
func defaultDBPath() string {
if env := os.Getenv("STOCK_DB_PATH"); env != "" {
return env
}
exe, err := os.Executable()
if err != nil {
return "stock.db"
}
return filepath.Join(filepath.Dir(exe), "stock.db")
}