From ba322e3950f371d5f275dc9c81b1e149e2270287 Mon Sep 17 00:00:00 2001 From: strange Date: Mon, 24 Nov 2025 18:29:59 +0800 Subject: [PATCH] pg batch write, boll indicator --- README.md | 6 +- api/pub.proto | 7 ++ api/trading.proto | 31 +++++++- cmd/trading/main.go | 22 +++--- config/exchange.toml | 3 +- go.mod | 11 +-- go.sum | 44 +++-------- internal/trading/backtest/account.go | 1 + internal/trading/backtest/types.go | 44 ++++++----- internal/trading/trading_data_persist.go | 31 +++++--- internal/trading/trading_grpc_server.go | 19 +++++ internal/trading/trading_service.go | 11 +++ pkg/config/config.go | 38 +++++++-- pkg/indicator/boll.go | 96 +++++++++++++++++++++++ pkg/indicator/indicator_registry.go | 3 + pkg/storage/ck/clickhouse_batch_writer.go | 61 +------------- pkg/storage/persist/persist.go | 61 ++++++++++++++ pkg/storage/persist/pg_batch_writer.go | 67 ++++++++++++++++ pkg/storage/persist/rdb.go | 16 ++++ 19 files changed, 426 insertions(+), 146 deletions(-) create mode 100644 pkg/indicator/boll.go create mode 100644 pkg/storage/persist/persist.go create mode 100644 pkg/storage/persist/pg_batch_writer.go diff --git a/README.md b/README.md index 712e6d5..f846c54 100644 --- a/README.md +++ b/README.md @@ -87,4 +87,8 @@ strategy0: 趋势追踪，增长趋势， 回测信号可视化, /trading/strategySeries 一样从postgres拉信号/订单数据 -go plugin插件化(指标/策略), docker中运行回测 +go plugin插件化(指标/策略), 调度docker容器运行回测, grpc资源访问 + +确定潜在的可调参数 + +Wasm(WebAssembly), 接口类型WIT diff --git a/api/pub.proto b/api/pub.proto index a30acb6..6e69062 100644 --- a/api/pub.proto +++ b/api/pub.proto @@ -149,3 +149,10 @@ message SeriesRange { uint32 windowExtra = 10; // 需要额外拉取更早的k线条数 uint32 limit = 11; // 大于0时检查, 数据长度超过limit则返回错误 } + +// 分页参数 +message Paging { + int32 page = 1; + int32 size = 2; + bool asc = 3; // 升序排序 +} diff --git a/api/trading.proto b/api/trading.proto index 5c88608..33e555a 100644 --- a/api/trading.proto +++ b/api/trading.proto @@ -8,7 +8,9 @@ option go_package = "./pb"; service TradingService { rpc IndicatorSeries(ReqIndicatorSeries) returns (RspIndicatorSeries); // 获取指标序列 rpc StrategySeries(ReqStrategySeries) returns (RspStrategySeries); // 获取指标序列 - rpc Backtest(ReqBacktest) returns (RspBacktest); // 交易计划回测 + rpc Backtest(ReqBacktest) returns (RspBacktest); // 交易计划回测 + rpc BacktestLog(ReqBacktestLog) returns (RspBacktestLog); // 交易计划回测记录 + rpc BacktestLogTrades(ReqBacktestLogTrades) returns (RspBacktestLogTrades); // 交易计划回测交易单详情 } message ReqIndicatorSeries { @@ -38,5 +40,32 @@ message ReqBacktest { string etime = 3; } message RspBacktest { + int64 backtestId = 1; +} + +message BacktestLog { + int64 backtestId = 1; + int64 planId = 2; + int64 stime = 3; + int64 etime = 4; + string interval = 5; // 交易周期 +} +message ReqBacktestLog { + Paging paging = 1; +} +message RspBacktestLog { + repeated BacktestLog logs = 1; +} + +message BacktestTrade { + int64 id = 1; + int64 ctime = 2; // 交易时间 +} +message ReqBacktestLogTrades { + Paging paging = 1; + int64 backtestId = 2; +} +message RspBacktestLogTrades { + repeated BacktestTrade trades = 1; } diff --git a/cmd/trading/main.go b/cmd/trading/main.go index e714798..c061f76 100644 --- a/cmd/trading/main.go +++ b/cmd/trading/main.go @@ -1,6 +1,7 @@ package main import ( + "context" "fmt" "net" "sig-pub/api/pb" @@ -10,7 +11,6 @@ import ( "sig-pub/pkg/grpc/discovery" "sig-pub/pkg/grpc/interceptor" "sig-pub/pkg/mq" - "sig-pub/pkg/storage/ck" "sig-pub/pkg/storage/persist" "sig-pub/pkg/utils/exit" "sig-pub/pkg/zlog" @@ -56,16 +56,20 @@ func main() { if err := rdb.Init(); err != nil { panic(err) } - // clickhouse - ckBatchWriter := ck.NewClickhouseBatchWriter(conf.Database.Clickhouse) - if err := ckBatchWriter.Init(); err != nil { - panic(err) - } - ckDB := ck.NewClickhouseDB(conf.Database.Clickhouse) - if err := ckDB.Init(); err != nil { + pgBatchWriter := persist.NewPGBatchWriter() + if err := pgBatchWriter.Init(context.Background(), conf.Database.Postgres.ConnString()); err != nil { panic(err) } - tradingDataPersist := trading.NewTradingDataPersist(rdb, ckDB, ckBatchWriter) + // clickhouse + // ckBatchWriter := ck.NewClickhouseBatchWriter(conf.Database.Clickhouse) + // if err := ckBatchWriter.Init(); err != nil { + // panic(err) + // } + // ckDB := ck.NewClickhouseDB(conf.Database.Clickhouse) + // if err := ckDB.Init(); err != nil { + // panic(err) + // } + tradingDataPersist := trading.NewTradingDataPersist(rdb, pgBatchWriter, nil, nil) if err := tradingDataPersist.Init(); err != nil { panic(err) } diff --git a/config/exchange.toml b/config/exchange.toml index 07f7bca..e65fb93 100644 --- a/config/exchange.toml +++ b/config/exchange.toml @@ -18,7 +18,8 @@ receiveBuffer = 4096 marketSubscribeLimit = 16 consumeBatch = 1024 consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume -httpProxy = "http://192.168.1.5:7890" +httpProxy = "" +# httpProxy = "http://192.168.1.5:7890" # httpProxy = "http://10.255.183.209:7890" # 模拟盘API交易地址如下： diff --git a/go.mod b/go.mod index 0a5e64f..2551535 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ go 1.24.0 toolchain go1.24.7 require ( - github.com/VictoriaMetrics/metrics v1.36.0 + github.com/ClickHouse/clickhouse-go/v2 v2.40.3 github.com/bwmarrin/snowflake v0.3.0 github.com/bytedance/sonic v1.13.2 github.com/dsnet/golib/unitconv v1.0.2 @@ -17,6 +17,7 @@ require ( github.com/govalues/decimal v0.1.36 github.com/hashicorp/consul/api v1.32.1 github.com/influxdata/influxdb-client-go/v2 v2.14.0 + github.com/jackc/pgx/v5 v5.7.6 github.com/jhump/protoreflect/v2 v2.0.0-beta.2 github.com/klauspost/compress v1.18.0 github.com/lib/pq v1.10.9 @@ -29,6 +30,7 @@ require ( go.etcd.io/etcd/api/v3 v3.6.1 go.etcd.io/etcd/client/v3 v3.6.1 go.uber.org/zap v1.27.0 + golang.org/x/crypto v0.42.0 golang.org/x/net v0.44.0 golang.org/x/sync v0.17.0 golang.org/x/time v0.8.0 @@ -36,6 +38,7 @@ require ( google.golang.org/grpc v1.71.1 google.golang.org/protobuf v1.36.6 gopkg.in/natefinch/lumberjack.v2 v2.2.1 + gorm.io/driver/clickhouse v0.7.0 gorm.io/driver/mysql v1.5.7 gorm.io/driver/postgres v1.6.0 gorm.io/gorm v1.30.0 @@ -43,7 +46,6 @@ require ( require ( github.com/ClickHouse/ch-go v0.68.0 // indirect - github.com/ClickHouse/clickhouse-go/v2 v2.40.3 // indirect github.com/andybalholm/brotli v1.2.0 // indirect github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect github.com/armon/go-metrics v0.4.1 // indirect @@ -81,7 +83,6 @@ require ( github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect - github.com/jackc/pgx/v5 v5.6.0 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect @@ -109,20 +110,16 @@ require ( github.com/subosito/gotenv v1.6.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.12 // indirect - github.com/valyala/fastrand v1.1.0 // indirect - github.com/valyala/histogram v1.2.0 // indirect go.etcd.io/etcd/client/pkg/v3 v3.6.1 // indirect go.opentelemetry.io/otel v1.38.0 // indirect go.opentelemetry.io/otel/trace v1.38.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/arch v0.15.0 // indirect - golang.org/x/crypto v0.42.0 // indirect golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 // indirect golang.org/x/sys v0.36.0 // indirect golang.org/x/text v0.29.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20250303144028-a0af3efb3deb // indirect gopkg.in/yaml.v3 v3.0.1 // indirect - gorm.io/driver/clickhouse v0.7.0 // indirect ) diff --git a/go.sum b/go.sum index 15eb908..470f2ee 100644 --- a/go.sum +++ b/go.sum @@ -4,8 +4,6 @@ github.com/ClickHouse/clickhouse-go/v2 v2.40.3 h1:46jB4kKwVDUOnECpStKMVXxvR0Cg9z github.com/ClickHouse/clickhouse-go/v2 v2.40.3/go.mod h1:qO0HwvjCnTB4BPL/k6EE3l4d9f/uF+aoimAhJX70eKA= github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ= github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk= -github.com/VictoriaMetrics/metrics v1.36.0 h1:f3SZMpLgIG4hJm2zfDs6wicxQ/QNWBZekY5rEGgbHKs= -github.com/VictoriaMetrics/metrics v1.36.0/go.mod h1:r7hveu6xMdUACXvB8TYdAj8WEsKzWB0EkpJN+RDtOf8= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= @@ -83,9 +81,8 @@ github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2 github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE= github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk= -github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= -github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s= @@ -170,8 +167,6 @@ github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/b github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8= github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= -github.com/hashicorp/go-version v1.2.1 h1:zEfKbn2+PDgroKdiOzqiE8rsmLqU2uwi5PB5pBJ3TkI= -github.com/hashicorp/go-version v1.2.1/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= github.com/hashicorp/go-version v1.7.0 h1:5tqGy27NaOTB8yJKUZELlFAS/LTKJkrmONwQKeRZfjY= github.com/hashicorp/go-version v1.7.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= @@ -191,8 +186,8 @@ github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsI github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= -github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY= -github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw= +github.com/jackc/pgx/v5 v5.7.6 h1:rWQc5FwZSPX58r1OQmkuaNicxdmExaEz5A2DO2hUuTk= +github.com/jackc/pgx/v5 v5.7.6/go.mod h1:aruU7o91Tc2q2cFp5h4uP3f6ztExVpyVv88Xl/8Vl8M= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/jhump/protoreflect/v2 v2.0.0-beta.2 h1:qZU+rEZUOYTz1Bnhi3xbwn+VxdXkLVeEpAeZzVXLY88= @@ -347,9 +342,9 @@ github.com/stretchr/testify v1.7.2/go.mod h1:R6va5+xMeoiuVRoj+gSkQ7d3FALtqAAGI1F github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= -github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk= @@ -358,13 +353,11 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE= github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg= -github.com/valyala/fastrand v1.1.0 h1:f+5HkLW4rsgzdNoleUOB69hyT9IlD2ZQh9GyDMfb5G8= -github.com/valyala/fastrand v1.1.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ= -github.com/valyala/histogram v1.2.0 h1:wyYGAZZt3CpwUiIb9AU/Zbllg1llXyrtApRS815OLoQ= -github.com/valyala/histogram v1.2.0/go.mod h1:Hb4kBwb4UxsaNbbbh+RRz8ZR6pdodR57tzWUS3BUzXY= github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= github.com/xdg-go/scram v1.1.1/go.mod h1:RaEWvsqvNKKvBPvcKeFjrG2cJqOkHTiyTpzz23ni57g= github.com/xdg-go/stringprep v1.0.3/go.mod h1:W3f5j4i+9rC0kuIEJL0ky1VpHXQU3ocBgklLGvcBnW8= +github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= +github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= @@ -377,19 +370,14 @@ go.etcd.io/etcd/client/v3 v3.6.1/go.mod h1:fCbPUdjWNLfx1A6ATo9syUmFVxqHH9bCnPLBZ go.mongodb.org/mongo-driver v1.11.4/go.mod h1:PTSz5yu21bkT/wXpkS7WR5f0ddqw5quethTUn9WM+2g= go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= -go.opentelemetry.io/otel v1.34.0 h1:zRLXxLCgL1WyKsPVrgbSdMN4c0FMkDAskSTQP+0hdUY= -go.opentelemetry.io/otel v1.34.0/go.mod h1:OWFPOQ+h4G8xpyjgqo4SxJYdDQ/qmRH+wivy7zzx9oI= go.opentelemetry.io/otel v1.38.0 h1:RkfdswUDRimDg0m2Az18RKOsnI8UDzppJAtj01/Ymk8= go.opentelemetry.io/otel v1.38.0/go.mod h1:zcmtmQ1+YmQM9wrNsTGV/q/uyusom3P8RxwExxkZhjM= -go.opentelemetry.io/otel/metric v1.34.0 h1:+eTR3U0MyfWjRDhmFMxe2SsW64QrZ84AOhvqS7Y+PoQ= -go.opentelemetry.io/otel/metric v1.34.0/go.mod h1:CEDrp0fy2D0MvkXE+dPV7cMi8tWZwX3dmaIhwPOaqHE= go.opentelemetry.io/otel/metric v1.38.0 h1:Kl6lzIYGAh5M159u9NgiRkmoMKjvbsKtYRwgfrA6WpA= -go.opentelemetry.io/otel/sdk v1.34.0 h1:95zS4k/2GOy069d321O8jWgYsW3MzVV+KuSPKp7Wr1A= -go.opentelemetry.io/otel/sdk v1.34.0/go.mod h1:0e/pNiaMAqaykJGKbi+tSjWfNNHMTxoC9qANsCzbyxU= +go.opentelemetry.io/otel/metric v1.38.0/go.mod h1:kB5n/QoRM8YwmUahxvI3bO34eVtQf2i4utNVLr9gEmI= +go.opentelemetry.io/otel/sdk v1.37.0 h1:ItB0QUqnjesGRvNcmAcU0LyvkVyGJ2xftD29bWdDvKI= +go.opentelemetry.io/otel/sdk v1.37.0/go.mod h1:VredYzxUvuo2q3WRcDnKDjbdvmO0sCzOvVAiY+yUkAg= go.opentelemetry.io/otel/sdk/metric v1.34.0 h1:5CeK9ujjbFVL5c1PhLuStg1wxA7vQv7ce1EK0Gyvahk= go.opentelemetry.io/otel/sdk/metric v1.34.0/go.mod h1:jQ/r8Ze28zRKoNRdkjCZxfs6YvBTG1+YIqyFVFYec5w= -go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC8mh/k= -go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE= go.opentelemetry.io/otel/trace v1.38.0 h1:Fxk5bKrDZJUH+AMyyIXGcFAPah0oRcT+LuNtJrmcNLE= go.opentelemetry.io/otel/trace v1.38.0/go.mod h1:j1P9ivuFsTceSWe1oY+EeW3sc+Pp42sO++GHkg4wwhs= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= @@ -408,8 +396,6 @@ golang.org/x/crypto v0.0.0-20190923035154-9ee001bba392/go.mod h1:/lpIB1dKB+9EgE3 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= -golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE= -golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc= golang.org/x/crypto v0.42.0 h1:chiH31gIWm57EkTXpwnqf8qeuMUi0yekh6mT2AvFlqI= golang.org/x/crypto v0.42.0/go.mod h1:4+rDnOTJhQCx2q7/j6rAN5XDw8kPjeaXEUR2eL94ix8= golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 h1:nDVHiLt8aIbd/VzvPWN6kSOPE7+F/fNFDSXLVYkE/Iw= @@ -426,8 +412,6 @@ golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwY golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1/go.mod h1:9tjilg8BloeKEkVJvy7fQ90B1CfIiPueXVOjqfkSzI8= golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.38.0 h1:vRMAPTMaeGqVhG5QyLJHqNDwecKTomGeqbnfZyKlBI8= -golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8= golang.org/x/net v0.44.0 h1:evd8IRDyfNBMBTTY5XRF1vaZlD+EmWx6x8PkhR04H/I= golang.org/x/net v0.44.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -436,8 +420,6 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610= -golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -465,18 +447,16 @@ golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20= -golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/sys v0.36.0 h1:KVRy2GtZBrk1cBYA7MKu5bEZFxQk4NIDV6RLVcC8o0k= golang.org/x/sys v0.36.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.35.0 h1:bZBVKBudEyhRcajGcNc3jIfWPqV4y/Kt2XcoigOWtDQ= +golang.org/x/term v0.35.0/go.mod h1:TPGtkTLesOwf2DE8CgVYiZinHAOuy5AYUYT1lENIZnA= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= -golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0= -golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU= golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4= golang.org/x/time v0.8.0 h1:9i3RxcPv3PZnitoVGMPDKZSq1xW1gK1Xy3ArNOGZfEg= @@ -524,8 +504,6 @@ gorm.io/driver/mysql v1.5.7/go.mod h1:sEtPWMiqiN1N1cMXoXmBbd8C6/l+TESwriotuRRpkD gorm.io/driver/postgres v1.6.0 h1:2dxzU8xJ+ivvqTRph34QX+WrRaJlmfyPqXmoGVjMBa4= gorm.io/driver/postgres v1.6.0/go.mod h1:vUw0mrGgrTK+uPHEhAdV4sfFELrByKVGnaVRkXDhtWo= gorm.io/gorm v1.25.7/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8= -gorm.io/gorm v1.25.12 h1:I0u8i2hWQItBq1WfE0o2+WuL9+8L21K9e2HHSTE/0f8= -gorm.io/gorm v1.25.12/go.mod h1:xh7N7RHfYlNc5EmcI/El95gXusucDrQnHXe0+CgWcLQ= gorm.io/gorm v1.30.0 h1:qbT5aPv1UH8gI99OsRlvDToLxW5zR7FzS9acZDOZcgs= gorm.io/gorm v1.30.0/go.mod h1:8Z33v652h4//uMA76KjeDH8mJXPm1QNCYrMeatR0DOE= nullprogram.com/x/optparse v1.0.0/go.mod h1:KdyPE+Igbe0jQUrVfMqDMeJQIJZEuyV7pjYmp6pbG50= diff --git a/internal/trading/backtest/account.go b/internal/trading/backtest/account.go index be40f80..18b4aaa 100644 --- a/internal/trading/backtest/account.go +++ b/internal/trading/backtest/account.go @@ -267,6 +267,7 @@ func (a *BacktestAccount) ClosePosition(pos *trade.Position, kline types.Kline, trade.Pnl = profit trade.HoldTime = conver.TimeDurationFormat(time.Duration(trade.CloseTime-trade.Time)*time.Millisecond, ".") trade.PeakPx = pos.PeakPx + trade.Cash = a.CurrentEquity(closePrice) } // 记录最大回撤 [high, low, high, low] diff --git a/internal/trading/backtest/types.go b/internal/trading/backtest/types.go index e8dd789..f38bf82 100644 --- a/internal/trading/backtest/types.go +++ b/internal/trading/backtest/types.go @@ -77,7 +77,7 @@ type TradeStat struct { // BacktestTradingPlan 交易计划回测结果 type BacktestTradingPlan struct { - Id int64 `json:"id" gorm:"column:id"` // 测试id + Id int64 `json:"id" gorm:"column:id;primaryKey"` // 测试id UserId int64 `json:"userId" gorm:"column:user_id"` // 用户id PlanId int64 `json:"planId" gorm:"column:plan_id"` // 交易计划id InstId string `json:"instId" gorm:"column:inst_id"` // 交易产品id @@ -100,31 +100,33 @@ type BacktestTradingPlan struct { } func (BacktestTradingPlan) TableName() string { - return "backtest_trading_plan" + return "t_backtest_trading_plan" } // Trade 交易计划回测单 type Trade struct { - Id int64 `json:"id" gorm:"column:id"` // 交易id - BacktestId int64 `json:"backtestId" gorm:"column:backtest_id"` // 回测id - Ctime int64 `json:"ctime" gorm:"column:ctime"` // 创建时间 - Side types.Side `json:"side" gorm:"column:side"` // 交易方向 - Qty float64 `json:"qty" gorm:"column:qty"` // 交易量 - Price float64 `json:"price" gorm:"column:price"` // 开仓价格 - Fee float64 `json:"fee" gorm:"column:fee"` // 开仓手续费 - Leverage int32 `json:"leverage" gorm:"column:leverage"` // 杠杆倍数 - Time int64 `json:"time" gorm:"column:time"` // 开仓时间 - TimeK int64 `json:"timeK" gorm:"column:time_k"` // 开仓K线时间 - ClosePrice float64 `json:"closePrice" gorm:"column:close_price"` // 平仓价格 - CloseFee float64 `json:"closeFee" gorm:"column:close_fee"` // 平仓手续费 - CloseTime int64 `json:"closeTime" gorm:"column:close_time"` // 平仓时间 - CloseTimeK int64 `json:"closeTimeK" gorm:"column:close_time_k"` // 平仓K线时间 - CloseCause trade.Cause `json:"closeCause" gorm:"column:close_cause"` // 平仓原因 ["stoploss", "takeprofit", "trailing", "retrace", "signal"](“止损”、“止盈”、“动态跟踪”、“回撤”、“信号”) - Pnl float64 `json:"pnl" gorm:"column:pnl"` // 盈利/亏损 pnl = (t.ClosePrice-t.Price)*t.Qty - t.Fee - t.CloseFee - HoldTime string `json:"holdTime" gorm:"column:hold_time"` // 持仓时间 - PeakPx float64 `json:"peakPx" gorm:"column:peakPx"` // highest (for long) or lowest (for short) observed price since entry + Id int64 `json:"id" gorm:"column:id;primaryKey"` // 交易id + BacktestId int64 `json:"backtestId" gorm:"column:backtest_id;primaryKey"` // 回测id + Ctime int64 `json:"ctime" gorm:"column:ctime"` // 创建时间 + Side types.Side `json:"side" gorm:"column:side"` // 交易方向 + Qty float64 `json:"qty" gorm:"column:qty"` // 交易量 + Price float64 `json:"price" gorm:"column:price"` // 开仓价格 + Fee float64 `json:"fee" gorm:"column:fee"` // 开仓手续费 + Leverage int32 `json:"leverage" gorm:"column:leverage"` // 杠杆倍数 + Time int64 `json:"time" gorm:"column:time"` // 开仓时间 + ClosePrice float64 `json:"closePrice" gorm:"column:close_price"` // 平仓价格 + CloseFee float64 `json:"closeFee" gorm:"column:close_fee"` // 平仓手续费 + CloseTime int64 `json:"closeTime" gorm:"column:close_time"` // 平仓时间 + CloseCause trade.Cause `json:"closeCause" gorm:"column:close_cause"` // 平仓原因 ["stoploss", "takeprofit", "trailing", "retrace", "signal"](“止损”、“止盈”、“动态跟踪”、“回撤”、“信号”) + Pnl float64 `json:"pnl" gorm:"column:pnl"` // 盈利/亏损 pnl = (t.ClosePrice-t.Price)*t.Qty - t.Fee - t.CloseFee + Cash float64 `json:"cash" gorm:"column:cash"` // 平仓后账户净值 + HoldTime string `json:"holdTime" gorm:"column:hold_time"` // 持仓时间 + PeakPx float64 `json:"peakPx" gorm:"column:peak_px"` // highest (for long) or lowest (for short) observed price since entry + + // TimeK int64 `json:"timeK" gorm:"column:time_k"` // 开仓K线时间 + // CloseTimeK int64 `json:"closeTimeK" gorm:"column:close_time_k"` // 平仓K线时间 } func (Trade) TableName() string { - return "backtest_trading_trade" + return "t_backtest_trading_trade" } diff --git a/internal/trading/trading_data_persist.go b/internal/trading/trading_data_persist.go index a846df1..c8fd69c 100644 --- a/internal/trading/trading_data_persist.go +++ b/internal/trading/trading_data_persist.go @@ -11,23 +11,26 @@ import ( type TradingDataPersist struct { db *persist.RDB + pgBatchWriter *persist.PGBatchWriter ckDB *ck.ClickhouseDB ckBatchWriter *ck.ClickhouseBatchWriter } -func NewTradingDataPersist(db *persist.RDB, ckDB *ck.ClickhouseDB, ckBatchWriter *ck.ClickhouseBatchWriter) *TradingDataPersist { +func NewTradingDataPersist(db *persist.RDB, pgBatchWriter *persist.PGBatchWriter, + ckDB *ck.ClickhouseDB, ckBatchWriter *ck.ClickhouseBatchWriter) *TradingDataPersist { return &TradingDataPersist{ db: db, + pgBatchWriter: pgBatchWriter, ckDB: ckDB, ckBatchWriter: ckBatchWriter, } } func (p *TradingDataPersist) Init() (err error) { - err = p.ckDB.AutoMigrateTables( - &backtest.BacktestTradingPlan{}, - &backtest.Trade{}, - ) + // err = p.db.AutoMigrateTables( + // &backtest.BacktestTradingPlan{}, + // &backtest.Trade{}, + // ) return } @@ -46,10 +49,20 @@ func (p *TradingDataPersist) GetTradePlanById(planId int64) (plan *entity.TradeP // SaveBacktestTradingPlan 保存交易计划回测结果 func (p *TradingDataPersist) SaveBacktestTradingPlan(ctx context.Context, backtestTradingPlan *backtest.BacktestTradingPlan) (err error) { - err = ck.InsertBatch(p.ckBatchWriter, ctx, []*backtest.BacktestTradingPlan{backtestTradingPlan}) - if err != nil { - return + datas := make([]any, 0, len(backtestTradingPlan.Trades)+1) + datas = append(datas, backtestTradingPlan) + for _, trade := range backtestTradingPlan.Trades { + datas = append(datas, trade) } - err = ck.InsertBatch(p.ckBatchWriter, ctx, backtestTradingPlan.Trades) + + err = persist.InsertBatch(p.pgBatchWriter, ctx, datas) + return +} + +// ListBacktestLogs 用户交易计划回测记录查询 +func (p *TradingDataPersist) ListBacktestLogs(userId int64) (backtestLogs []*backtest.BacktestTradingPlan, err error) { + err = p.db.Select(&backtestLogs, ` + select * from t_backtest_trading_plan where user_id = ? order by id desc + `, userId) return } diff --git a/internal/trading/trading_grpc_server.go b/internal/trading/trading_grpc_server.go index 9b4fdeb..7ef4ae3 100644 --- a/internal/trading/trading_grpc_server.go +++ b/internal/trading/trading_grpc_server.go @@ -57,3 +57,22 @@ func (svr *TradingGrpcServer) Backtest(ctx context.Context, req *pb.ReqBacktest) rsp = new(pb.RspBacktest) return } + +// BacktestLog 回测记录查询 +func (svr *TradingGrpcServer) BacktestLog(ctx context.Context, req *pb.ReqBacktestLog) (rsp *pb.RspBacktestLog, err error) { + logs, err := svr.tradingService.BacktestLog(ctx, 10001) + if err != nil { + return + } + rsp = new(pb.RspBacktestLog) + for _, l := range logs { + log := &pb.BacktestLog{ + PlanId: l.PlanId, + Stime: l.SeriesBefore, + Etime: l.SeriesAfter, + BacktestId: l.Id, + } + rsp.Logs = append(rsp.Logs, log) + } + return +} diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 9a491a2..e8f5855 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -313,10 +313,21 @@ func (svc *TradingService) Backtest(ctx context.Context, planId, stime, etime in if err = tester.Init(10000, *plan); err != nil { return } + w := times.NewWatch() backtestTradingPlan, err := tester.Backtest(ctx, sr) if err != nil { return } + zlog.Infof("backtest use %s", w.ElapsedFmt("")) + w.Reset() err = svc.tradingDataPersist.SaveBacktestTradingPlan(ctx, backtestTradingPlan) + zlog.Infof("insert backtest ret use %s", w.ElapsedFmt("")) + return +} + +// BacktestLog 回测记录查询 +func (svc *TradingService) BacktestLog(ctx context.Context, userId int64) (backtestLogs []*backtest.BacktestTradingPlan, err error) { + // todo userid from ctx + backtestLogs, err = svc.tradingDataPersist.ListBacktestLogs(userId) return } diff --git a/pkg/config/config.go b/pkg/config/config.go index defa845..b2fad03 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -2,9 +2,11 @@ package config import ( "context" + "fmt" "net" "os" "sig-pub/pkg/zlog" + "strings" "time" "github.com/redis/go-redis/v9" @@ -102,11 +104,11 @@ type PostgresConfig struct { Postgres postgres.Config } -func (conf PostgresConfig) NewGormDB() (db *gorm.DB, err error) { +func (c PostgresConfig) NewGormDB() (db *gorm.DB, err error) { logLevel := logger.Silent - switch conf.LogMode { + switch c.LogMode { default: - zlog.Errorf("unknow mysql gorm logmode: %s", conf.LogMode) + zlog.Errorf("unknow mysql gorm logmode: %s", c.LogMode) case "": case "silent": logLevel = logger.Silent @@ -117,11 +119,37 @@ func (conf PostgresConfig) NewGormDB() (db *gorm.DB, err error) { case "info": logLevel = logger.Info } - conf.Logger = logger.Default.LogMode(logLevel) - db, err = gorm.Open(postgres.New(conf.Postgres), &conf) + c.Logger = logger.Default.LogMode(logLevel) + db, err = gorm.Open(postgres.New(c.Postgres), &c) return } +func (c PostgresConfig) ConnString() string { + // "host=127.0.0.1 port=5432 user=postgres password=123456 dbname=sig sslmode=disable TimeZone=Asia/Shanghai" + dsns := strings.Split(c.Postgres.DSN, " ") + var user, password, host, port, dbname string + for _, dsn := range dsns { + kv := strings.Split(dsn, "=") + if len(kv) != 2 { + continue + } + v := kv[1] + switch kv[0] { + case "user": + user = v + case "password": + password = v + case "host": + host = v + case "port": + port = v + case "dbname": + dbname = v + } + } + return fmt.Sprintf("postgres://%s:%s@%s:%s/%s?pool_min_conns=4&pool_max_conns=20", user, password, host, port, dbname) +} + type ClickhouseConfig struct { Logsql bool LogsqlCk bool diff --git a/pkg/indicator/boll.go b/pkg/indicator/boll.go new file mode 100644 index 0000000..ae13207 --- /dev/null +++ b/pkg/indicator/boll.go @@ -0,0 +1,96 @@ +package indicator + +import ( + "math" + "sig-pub/pkg/types" +) + +// BollMB 布林带中轨 +type BollMB struct { +} + +func (c *BollMB) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "BollMB", + Input: []types.InputArg{ + {Name: "window", Type: types.InputTypeUInt, Desc: "窗口大小"}, + }, + } +} + +func (c *BollMB) CandlePeriods(ctx IIndicatorContext) int16 { + return ctx.Input().Int16("window") +} + +func (c *BollMB) Calculate(ctx IIndicatorContext) (vector float64) { + window := ctx.Input().Int16("window") + closeSeries := ctx.Series(0, int16(window)).Close() + vector = closeSeries.Avg() + return +} + +// BollUB 布林带上轨 +type BollUB struct { +} + +func (c *BollUB) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "BollUB", + Input: []types.InputArg{ + {Name: "window", Type: types.InputTypeUInt, Desc: "窗口大小"}, + }, + } +} + +func (c *BollUB) CandlePeriods(ctx IIndicatorContext) int16 { + return ctx.Input().Int16("window") +} + +func (c *BollUB) Calculate(ctx IIndicatorContext) (vector float64) { + window := ctx.Input().Int16("window") + closeSeries := ctx.Series(0, int16(window)).Close() + mb := closeSeries.Avg() + + // 标准差σ_t = sqrt(∑(P-MB)^2 / (n-1)) + sst := float64(0) + for _, p := range closeSeries { + sst += math.Pow(p-mb, 2) + } + sigma := math.Sqrt(sst / float64(window-1)) + + vector = mb + 2*sigma + return +} + +// BollLB 布林带下轨 +type BollLB struct { +} + +func (c *BollLB) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "BollLB", + Input: []types.InputArg{ + {Name: "window", Type: types.InputTypeUInt, Desc: "窗口大小"}, + }, + } +} + +func (c *BollLB) CandlePeriods(ctx IIndicatorContext) int16 { + return ctx.Input().Int16("window") +} + +func (c *BollLB) Calculate(ctx IIndicatorContext) (vector float64) { + window := ctx.Input().Int16("window") + closeSeries := ctx.Series(0, int16(window)).Close() + mb := closeSeries.Avg() + + // 标准差σ_t = sqrt(∑(P-MB)^2 / (n-1)) + sst := float64(0) + for _, p := range closeSeries { + sst += math.Pow(p-mb, 2) + } + sigma := math.Sqrt(sst / float64(window-1)) + + vector = mb - 2*sigma + return +} diff --git a/pkg/indicator/indicator_registry.go b/pkg/indicator/indicator_registry.go index ac53b08..8de38f7 100644 --- a/pkg/indicator/indicator_registry.go +++ b/pkg/indicator/indicator_registry.go @@ -27,6 +27,9 @@ func (r *IndicatorRegistry) Init() (err error) { r.MustRegistIndicatorW(&Macd{}) r.MustRegistIndicatorW(&OBV{}) r.MustRegistIndicatorW(&WOBV{}) + r.MustRegistIndicatorW(&BollMB{}) + r.MustRegistIndicatorW(&BollUB{}) + r.MustRegistIndicatorW(&BollLB{}) return } diff --git a/pkg/storage/ck/clickhouse_batch_writer.go b/pkg/storage/ck/clickhouse_batch_writer.go index e77f326..f064dd8 100644 --- a/pkg/storage/ck/clickhouse_batch_writer.go +++ b/pkg/storage/ck/clickhouse_batch_writer.go @@ -3,9 +3,8 @@ package ck import ( "context" "fmt" - "reflect" "sig-pub/pkg/config" - "strings" + "sig-pub/pkg/storage/persist" "time" "github.com/ClickHouse/clickhouse-go/v2/lib/driver" @@ -46,62 +45,6 @@ func (c *ClickhouseBatchWriter) Init() (err error) { return } -func reflectGormData(data any) (table string, columns []string, values map[string]any, err error) { - // value := reflect.ValueOf(data) - // tableNameM := value.MethodByName("TableName") - // _ = tableNameM.Call(nil) - defer func() { - if r := recover(); r != nil { - err = fmt.Errorf("%T parse gorm data error %v", data, r) - } - }() - - values = make(map[string]any) - refValue := reflect.ValueOf(data) - refType := reflect.TypeOf(data) - - tableNameM := refValue.MethodByName("TableName") - if !tableNameM.IsValid() { - err = fmt.Errorf("type %T not has method TableName", data) - return - } - rsp := tableNameM.Call(nil) - table = rsp[0].Interface().(string) - - if refValue.Kind() == reflect.Ptr { - refValue = refValue.Elem() - } - if refType.Kind() == reflect.Ptr { - refType = refType.Elem() - } - - for i := 0; i < refType.NumField(); i++ { - field := refType.Field(i) - value := refValue.Field(i).Interface() - // value - tag := field.Tag.Get("gorm") - column := getGormTagColumnName(tag) - if column == "" || strings.HasPrefix(column, "-") { // 忽略字段 - continue - } - columns = append(columns, column) - values[column] = value - } - return -} - -func getGormTagColumnName(tag string) (column string) { - i1 := strings.Index(tag, "column:") - if i1 < 0 { - return - } - i2 := strings.Index(tag[i1+7:], ";") - if i2 < 0 { - return tag[i1+7:] - } - return tag[i1+7 : i2+7] -} - // InsertBatch 批量插入gorm标签的结构体, 结构体用指针！ func InsertBatch[T any](c *ClickhouseBatchWriter, ctx context.Context, datas []T) (err error) { if len(datas) == 0 { @@ -113,7 +56,7 @@ func InsertBatch[T any](c *ClickhouseBatchWriter, ctx context.Context, datas []T for _, data := range datas { // 优化原生批量插入: https://clickhouse.com/docs/en/integrations/go#batch-insert - table, columns, values, e := reflectGormData(data) + table, columns, values, e := persist.ReflectGormData(data) if e != nil { err = e return diff --git a/pkg/storage/persist/persist.go b/pkg/storage/persist/persist.go new file mode 100644 index 0000000..4eee9fa --- /dev/null +++ b/pkg/storage/persist/persist.go @@ -0,0 +1,61 @@ +package persist + +import ( + "fmt" + "reflect" + "strings" +) + +// ReflectGormData todo cache data type info +func ReflectGormData(data any) (table string, columns []string, values map[string]any, err error) { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("%T parse gorm data error %v", data, r) + } + }() + + values = make(map[string]any) + refValue := reflect.ValueOf(data) + refType := reflect.TypeOf(data) + + tableNameM := refValue.MethodByName("TableName") + if !tableNameM.IsValid() { + err = fmt.Errorf("type %T not has method TableName", data) + return + } + rsp := tableNameM.Call(nil) + table = rsp[0].Interface().(string) + + if refValue.Kind() == reflect.Ptr { + refValue = refValue.Elem() + } + if refType.Kind() == reflect.Ptr { + refType = refType.Elem() + } + + for i := 0; i < refType.NumField(); i++ { + field := refType.Field(i) + value := refValue.Field(i).Interface() + // value + tag := field.Tag.Get("gorm") + column := getGormTagColumnName(tag) + if column == "" || strings.HasPrefix(column, "-") { // 忽略字段 + continue + } + columns = append(columns, column) + values[column] = value + } + return +} + +func getGormTagColumnName(tag string) (column string) { + i1 := strings.Index(tag, "column:") + if i1 < 0 { + return + } + i2 := strings.Index(tag[i1+7:], ";") + if i2 < 0 { + return tag[i1+7:] + } + return tag[i1+7 : i2+7] +} diff --git a/pkg/storage/persist/pg_batch_writer.go b/pkg/storage/persist/pg_batch_writer.go new file mode 100644 index 0000000..cb6655b --- /dev/null +++ b/pkg/storage/persist/pg_batch_writer.go @@ -0,0 +1,67 @@ +package persist + +import ( + "context" + "sig-pub/pkg/utils/exit" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +type PGBatchWriter struct { + pool *pgxpool.Pool +} + +func NewPGBatchWriter() *PGBatchWriter { + return &PGBatchWriter{} +} + +func (w *PGBatchWriter) Init(ctx context.Context, connString string) (err error) { + w.pool, err = pgxpool.New(ctx, connString) + if err != nil { + return + } + exit.AddHook(w.pool.Close, exit.WithOrderTail()) + + return +} + +func InsertBatch[T any](w *PGBatchWriter, ctx context.Context, datas []T) (err error) { + if len(datas) == 0 { + return + } + + tableColumns := make(map[string][]string, 3) + tableValues := make(map[string][][]any, 3) + for _, data := range datas { + table, columns, valuesM, e := ReflectGormData(data) + if e != nil { + err = e + return + } + if cs, ok := tableColumns[table]; !ok { + tableColumns[table] = columns + } else { + columns = cs + } + var values []any + for _, c := range columns { + v := valuesM[c] + values = append(values, v) + } + tableValues[table] = append(tableValues[table], values) + } + tx, err := w.pool.Begin(ctx) + if err != nil { + return + } + defer tx.Rollback(ctx) + for table, values := range tableValues { + _, err = tx.CopyFrom(ctx, pgx.Identifier{table}, tableColumns[table], pgx.CopyFromRows(values)) + if err != nil { + return + } + } + err = tx.Commit(ctx) + return +} diff --git a/pkg/storage/persist/rdb.go b/pkg/storage/persist/rdb.go index 22ac7ad..6247556 100644 --- a/pkg/storage/persist/rdb.go +++ b/pkg/storage/persist/rdb.go @@ -1,6 +1,8 @@ package persist import ( + "sig-pub/pkg/zlog" + "gorm.io/gorm" ) @@ -41,3 +43,17 @@ func (m *RDB) UpdateBy(data any) (rowsAffected int64, err error) { rowsAffected, err = tx.RowsAffected, tx.Error return } + +// AutoMigrateTables 自动对齐表结构，自动根据字段修改数据库表结构，只会加改不会删字段 +func (m *RDB) AutoMigrateTables(gormStructs ...any) (err error) { + if len(gormStructs) == 0 { + return + } + err = m.db.AutoMigrate(gormStructs...) + if err != nil { + zlog.Errorf("type auto migrate postgres table error: %v", err) + return + } + // db.Set("gorm:table_options", "ENGINE=Distributed(cluster, default, hits)").AutoMigrate(&entity.TradeRecord{}) + return +}