Merge remote-tracking branch 'refs/remotes/origin/main'

This commit is contained in:
王性驊 2025-08-20 20:35:40 +08:00
commit e36950e6b3
11 changed files with 278 additions and 52 deletions

View File

@ -1,5 +1,5 @@
// Code generated by goctl. DO NOT EDIT.
// goctl 1.8.1
// goctl 1.8.5
// Source: blockchain.proto
package blockchainservice
@ -15,6 +15,7 @@ import (
type (
HistoryReq = app_cloudep_blockchain.HistoryReq
ListCandleDataResp = app_cloudep_blockchain.ListCandleDataResp
ListSymbolsRequest = app_cloudep_blockchain.ListSymbolsRequest
ListSymbolsResponse = app_cloudep_blockchain.ListSymbolsResponse
NoneReq = app_cloudep_blockchain.NoneReq
@ -25,6 +26,7 @@ type (
// ListSymbols retrieves all available trading symbols.
ListSymbols(ctx context.Context, in *ListSymbolsRequest, opts ...grpc.CallOption) (*ListSymbolsResponse, error)
GetHistoryKlineData(ctx context.Context, in *HistoryReq, opts ...grpc.CallOption) (*OKResp, error)
ListCandleData(ctx context.Context, in *HistoryReq, opts ...grpc.CallOption) (*ListCandleDataResp, error)
// Ping is a health-check endpoint.
Ping(ctx context.Context, in *NoneReq, opts ...grpc.CallOption) (*OKResp, error)
}
@ -51,6 +53,11 @@ func (m *defaultBlockchainService) GetHistoryKlineData(ctx context.Context, in *
return client.GetHistoryKlineData(ctx, in, opts...)
}
func (m *defaultBlockchainService) ListCandleData(ctx context.Context, in *HistoryReq, opts ...grpc.CallOption) (*ListCandleDataResp, error) {
client := app_cloudep_blockchain.NewBlockchainServiceClient(m.cli.Conn())
return client.ListCandleData(ctx, in, opts...)
}
// Ping is a health-check endpoint.
func (m *defaultBlockchainService) Ping(ctx context.Context, in *NoneReq, opts ...grpc.CallOption) (*OKResp, error) {
client := app_cloudep_blockchain.NewBlockchainServiceClient(m.cli.Conn())

View File

@ -12,14 +12,14 @@ Binance:
WorkerSize: 20
RedisCluster:
Host: 10.0.0.13:6379
Host: localhost:6379
Type: node
Cassandra:
Hosts:
- 10.0.0.13
- localhost
Port: 9042
Keyspace: digimon
Keyspace: kline
UseAuth: true
Username: cassandra
Password: cassandra

View File

@ -1,6 +1,6 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.1
// protoc-gen-go v1.36.5
// protoc v3.19.4
// source: generate/rpc/blockchain.proto
@ -11,6 +11,7 @@ import (
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
unsafe "unsafe"
)
const (
@ -330,9 +331,45 @@ func (x *HistoryReq) GetEndTime() string {
return ""
}
type ListCandleDataResp struct {
state protoimpl.MessageState `protogen:"open.v1"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ListCandleDataResp) Reset() {
*x = ListCandleDataResp{}
mi := &file_generate_rpc_blockchain_proto_msgTypes[6]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *ListCandleDataResp) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ListCandleDataResp) ProtoMessage() {}
func (x *ListCandleDataResp) ProtoReflect() protoreflect.Message {
mi := &file_generate_rpc_blockchain_proto_msgTypes[6]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use ListCandleDataResp.ProtoReflect.Descriptor instead.
func (*ListCandleDataResp) Descriptor() ([]byte, []int) {
return file_generate_rpc_blockchain_proto_rawDescGZIP(), []int{6}
}
var File_generate_rpc_blockchain_proto protoreflect.FileDescriptor
var file_generate_rpc_blockchain_proto_rawDesc = []byte{
var file_generate_rpc_blockchain_proto_rawDesc = string([]byte{
0x0a, 0x1d, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x2f, 0x72, 0x70, 0x63, 0x2f, 0x62,
0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12,
0x0a, 0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x22, 0x08, 0x0a, 0x06, 0x4f,
@ -364,17 +401,23 @@ var file_generate_rpc_blockchain_proto_rawDesc = []byte{
0x0a, 0x0a, 0x73, 0x74, 0x61, 0x72, 0x74, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01,
0x28, 0x09, 0x52, 0x09, 0x73, 0x74, 0x61, 0x72, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x19, 0x0a,
0x08, 0x65, 0x6e, 0x64, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52,
0x07, 0x65, 0x6e, 0x64, 0x54, 0x69, 0x6d, 0x65, 0x32, 0xd7, 0x01, 0x0a, 0x11, 0x42, 0x6c, 0x6f,
0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x4e,
0x0a, 0x0b, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x79, 0x6d, 0x62, 0x6f, 0x6c, 0x73, 0x12, 0x1e, 0x2e,
0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x53,
0x79, 0x6d, 0x62, 0x6f, 0x6c, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1f, 0x2e,
0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x53,
0x79, 0x6d, 0x62, 0x6f, 0x6c, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x41,
0x0a, 0x13, 0x47, 0x65, 0x74, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x4b, 0x6c, 0x69, 0x6e,
0x65, 0x44, 0x61, 0x74, 0x61, 0x12, 0x16, 0x2e, 0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61,
0x69, 0x6e, 0x2e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x52, 0x65, 0x71, 0x1a, 0x12, 0x2e,
0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x2e, 0x4f, 0x4b, 0x52, 0x65, 0x73,
0x07, 0x65, 0x6e, 0x64, 0x54, 0x69, 0x6d, 0x65, 0x22, 0x14, 0x0a, 0x12, 0x4c, 0x69, 0x73, 0x74,
0x43, 0x61, 0x6e, 0x64, 0x6c, 0x65, 0x44, 0x61, 0x74, 0x61, 0x52, 0x65, 0x73, 0x70, 0x32, 0xa1,
0x02, 0x0a, 0x11, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x53, 0x65, 0x72,
0x76, 0x69, 0x63, 0x65, 0x12, 0x4e, 0x0a, 0x0b, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x79, 0x6d, 0x62,
0x6f, 0x6c, 0x73, 0x12, 0x1e, 0x2e, 0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e,
0x2e, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x79, 0x6d, 0x62, 0x6f, 0x6c, 0x73, 0x52, 0x65, 0x71, 0x75,
0x65, 0x73, 0x74, 0x1a, 0x1f, 0x2e, 0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e,
0x2e, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x79, 0x6d, 0x62, 0x6f, 0x6c, 0x73, 0x52, 0x65, 0x73, 0x70,
0x6f, 0x6e, 0x73, 0x65, 0x12, 0x41, 0x0a, 0x13, 0x47, 0x65, 0x74, 0x48, 0x69, 0x73, 0x74, 0x6f,
0x72, 0x79, 0x4b, 0x6c, 0x69, 0x6e, 0x65, 0x44, 0x61, 0x74, 0x61, 0x12, 0x16, 0x2e, 0x62, 0x6c,
0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x2e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79,
0x52, 0x65, 0x71, 0x1a, 0x12, 0x2e, 0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e,
0x2e, 0x4f, 0x4b, 0x52, 0x65, 0x73, 0x70, 0x12, 0x48, 0x0a, 0x0e, 0x4c, 0x69, 0x73, 0x74, 0x43,
0x61, 0x6e, 0x64, 0x6c, 0x65, 0x44, 0x61, 0x74, 0x61, 0x12, 0x16, 0x2e, 0x62, 0x6c, 0x6f, 0x63,
0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x2e, 0x48, 0x69, 0x73, 0x74, 0x6f, 0x72, 0x79, 0x52, 0x65,
0x71, 0x1a, 0x1e, 0x2e, 0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x2e, 0x4c,
0x69, 0x73, 0x74, 0x43, 0x61, 0x6e, 0x64, 0x6c, 0x65, 0x44, 0x61, 0x74, 0x61, 0x52, 0x65, 0x73,
0x70, 0x12, 0x2f, 0x0a, 0x04, 0x50, 0x69, 0x6e, 0x67, 0x12, 0x13, 0x2e, 0x62, 0x6c, 0x6f, 0x63,
0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x2e, 0x4e, 0x6f, 0x6e, 0x65, 0x52, 0x65, 0x71, 0x1a, 0x12,
0x2e, 0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61, 0x69, 0x6e, 0x2e, 0x4f, 0x4b, 0x52, 0x65,
@ -382,21 +425,21 @@ var file_generate_rpc_blockchain_proto_rawDesc = []byte{
0x6e, 0x65, 0x74, 0x2f, 0x64, 0x69, 0x67, 0x69, 0x6d, 0x6f, 0x6e, 0x2f, 0x61, 0x70, 0x70, 0x2d,
0x63, 0x6c, 0x6f, 0x75, 0x64, 0x65, 0x70, 0x2d, 0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x63, 0x68, 0x61,
0x69, 0x6e, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
})
var (
file_generate_rpc_blockchain_proto_rawDescOnce sync.Once
file_generate_rpc_blockchain_proto_rawDescData = file_generate_rpc_blockchain_proto_rawDesc
file_generate_rpc_blockchain_proto_rawDescData []byte
)
func file_generate_rpc_blockchain_proto_rawDescGZIP() []byte {
file_generate_rpc_blockchain_proto_rawDescOnce.Do(func() {
file_generate_rpc_blockchain_proto_rawDescData = protoimpl.X.CompressGZIP(file_generate_rpc_blockchain_proto_rawDescData)
file_generate_rpc_blockchain_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_generate_rpc_blockchain_proto_rawDesc), len(file_generate_rpc_blockchain_proto_rawDesc)))
})
return file_generate_rpc_blockchain_proto_rawDescData
}
var file_generate_rpc_blockchain_proto_msgTypes = make([]protoimpl.MessageInfo, 6)
var file_generate_rpc_blockchain_proto_msgTypes = make([]protoimpl.MessageInfo, 7)
var file_generate_rpc_blockchain_proto_goTypes = []any{
(*OKResp)(nil), // 0: blockchain.OKResp
(*NoneReq)(nil), // 1: blockchain.NoneReq
@ -404,17 +447,20 @@ var file_generate_rpc_blockchain_proto_goTypes = []any{
(*ListSymbolsResponse)(nil), // 3: blockchain.ListSymbolsResponse
(*Symbol)(nil), // 4: blockchain.Symbol
(*HistoryReq)(nil), // 5: blockchain.HistoryReq
(*ListCandleDataResp)(nil), // 6: blockchain.ListCandleDataResp
}
var file_generate_rpc_blockchain_proto_depIdxs = []int32{
4, // 0: blockchain.ListSymbolsResponse.symbols:type_name -> blockchain.Symbol
2, // 1: blockchain.BlockchainService.ListSymbols:input_type -> blockchain.ListSymbolsRequest
5, // 2: blockchain.BlockchainService.GetHistoryKlineData:input_type -> blockchain.HistoryReq
1, // 3: blockchain.BlockchainService.Ping:input_type -> blockchain.NoneReq
3, // 4: blockchain.BlockchainService.ListSymbols:output_type -> blockchain.ListSymbolsResponse
0, // 5: blockchain.BlockchainService.GetHistoryKlineData:output_type -> blockchain.OKResp
0, // 6: blockchain.BlockchainService.Ping:output_type -> blockchain.OKResp
4, // [4:7] is the sub-list for method output_type
1, // [1:4] is the sub-list for method input_type
5, // 3: blockchain.BlockchainService.ListCandleData:input_type -> blockchain.HistoryReq
1, // 4: blockchain.BlockchainService.Ping:input_type -> blockchain.NoneReq
3, // 5: blockchain.BlockchainService.ListSymbols:output_type -> blockchain.ListSymbolsResponse
0, // 6: blockchain.BlockchainService.GetHistoryKlineData:output_type -> blockchain.OKResp
6, // 7: blockchain.BlockchainService.ListCandleData:output_type -> blockchain.ListCandleDataResp
0, // 8: blockchain.BlockchainService.Ping:output_type -> blockchain.OKResp
5, // [5:9] is the sub-list for method output_type
1, // [1:5] is the sub-list for method input_type
1, // [1:1] is the sub-list for extension type_name
1, // [1:1] is the sub-list for extension extendee
0, // [0:1] is the sub-list for field type_name
@ -429,9 +475,9 @@ func file_generate_rpc_blockchain_proto_init() {
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_generate_rpc_blockchain_proto_rawDesc,
RawDescriptor: unsafe.Slice(unsafe.StringData(file_generate_rpc_blockchain_proto_rawDesc), len(file_generate_rpc_blockchain_proto_rawDesc)),
NumEnums: 0,
NumMessages: 6,
NumMessages: 7,
NumExtensions: 0,
NumServices: 1,
},
@ -440,7 +486,6 @@ func file_generate_rpc_blockchain_proto_init() {
MessageInfos: file_generate_rpc_blockchain_proto_msgTypes,
}.Build()
File_generate_rpc_blockchain_proto = out.File
file_generate_rpc_blockchain_proto_rawDesc = nil
file_generate_rpc_blockchain_proto_goTypes = nil
file_generate_rpc_blockchain_proto_depIdxs = nil
}

View File

@ -21,6 +21,7 @@ const _ = grpc.SupportPackageIsVersion9
const (
BlockchainService_ListSymbols_FullMethodName = "/blockchain.BlockchainService/ListSymbols"
BlockchainService_GetHistoryKlineData_FullMethodName = "/blockchain.BlockchainService/GetHistoryKlineData"
BlockchainService_ListCandleData_FullMethodName = "/blockchain.BlockchainService/ListCandleData"
BlockchainService_Ping_FullMethodName = "/blockchain.BlockchainService/Ping"
)
@ -31,6 +32,7 @@ type BlockchainServiceClient interface {
// ListSymbols retrieves all available trading symbols.
ListSymbols(ctx context.Context, in *ListSymbolsRequest, opts ...grpc.CallOption) (*ListSymbolsResponse, error)
GetHistoryKlineData(ctx context.Context, in *HistoryReq, opts ...grpc.CallOption) (*OKResp, error)
ListCandleData(ctx context.Context, in *HistoryReq, opts ...grpc.CallOption) (*ListCandleDataResp, error)
// Ping is a health-check endpoint.
Ping(ctx context.Context, in *NoneReq, opts ...grpc.CallOption) (*OKResp, error)
}
@ -63,6 +65,16 @@ func (c *blockchainServiceClient) GetHistoryKlineData(ctx context.Context, in *H
return out, nil
}
func (c *blockchainServiceClient) ListCandleData(ctx context.Context, in *HistoryReq, opts ...grpc.CallOption) (*ListCandleDataResp, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(ListCandleDataResp)
err := c.cc.Invoke(ctx, BlockchainService_ListCandleData_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *blockchainServiceClient) Ping(ctx context.Context, in *NoneReq, opts ...grpc.CallOption) (*OKResp, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(OKResp)
@ -80,6 +92,7 @@ type BlockchainServiceServer interface {
// ListSymbols retrieves all available trading symbols.
ListSymbols(context.Context, *ListSymbolsRequest) (*ListSymbolsResponse, error)
GetHistoryKlineData(context.Context, *HistoryReq) (*OKResp, error)
ListCandleData(context.Context, *HistoryReq) (*ListCandleDataResp, error)
// Ping is a health-check endpoint.
Ping(context.Context, *NoneReq) (*OKResp, error)
mustEmbedUnimplementedBlockchainServiceServer()
@ -98,6 +111,9 @@ func (UnimplementedBlockchainServiceServer) ListSymbols(context.Context, *ListSy
func (UnimplementedBlockchainServiceServer) GetHistoryKlineData(context.Context, *HistoryReq) (*OKResp, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetHistoryKlineData not implemented")
}
func (UnimplementedBlockchainServiceServer) ListCandleData(context.Context, *HistoryReq) (*ListCandleDataResp, error) {
return nil, status.Errorf(codes.Unimplemented, "method ListCandleData not implemented")
}
func (UnimplementedBlockchainServiceServer) Ping(context.Context, *NoneReq) (*OKResp, error) {
return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented")
}
@ -158,6 +174,24 @@ func _BlockchainService_GetHistoryKlineData_Handler(srv interface{}, ctx context
return interceptor(ctx, in, info, handler)
}
func _BlockchainService_ListCandleData_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(HistoryReq)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(BlockchainServiceServer).ListCandleData(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: BlockchainService_ListCandleData_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(BlockchainServiceServer).ListCandleData(ctx, req.(*HistoryReq))
}
return interceptor(ctx, in, info, handler)
}
func _BlockchainService_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(NoneReq)
if err := dec(in); err != nil {
@ -191,6 +225,10 @@ var BlockchainService_ServiceDesc = grpc.ServiceDesc{
MethodName: "GetHistoryKlineData",
Handler: _BlockchainService_GetHistoryKlineData_Handler,
},
{
MethodName: "ListCandleData",
Handler: _BlockchainService_ListCandleData_Handler,
},
{
MethodName: "Ping",
Handler: _BlockchainService_Ping_Handler,

View File

@ -33,10 +33,15 @@ message HistoryReq {
string end_time=4;
}
message ListCandleDataResp {
}
service BlockchainService{
// ListSymbols retrieves all available trading symbols.
rpc ListSymbols(ListSymbolsRequest) returns(ListSymbolsResponse);
rpc GetHistoryKlineData(HistoryReq)returns(OKResp);
rpc ListCandleData(HistoryReq)returns(ListCandleDataResp);
// Ping is a health-check endpoint.
rpc Ping(NoneReq) returns(OKResp);
}

View File

@ -3,6 +3,7 @@ package blockchainservicelogic
import (
"blockchain/internal/domain/usecase"
"context"
"fmt"
"time"
"blockchain/gen_result/pb/code.30cm.net/digimon/app-cloudep-blockchain"
@ -26,17 +27,81 @@ func NewGetHistoryKlineDataLogic(ctx context.Context, svcCtx *svc.ServiceContext
}
func (l *GetHistoryKlineDataLogic) GetHistoryKlineData(in *app_cloudep_blockchain.HistoryReq) (*app_cloudep_blockchain.OKResp, error) {
go func() {
st, _ := time.Parse(time.RFC3339, in.GetStartTime())
et, _ := time.Parse(time.RFC3339, in.GetEndTime())
// 1) 先驗證輸入(避免把錯誤時間丟進背景 goroutine
st, err := time.Parse(time.RFC3339, in.GetStartTime())
if err != nil {
return nil, fmt.Errorf("invalid startTime (RFC3339): %w", err)
}
et, err := time.Parse(time.RFC3339, in.GetEndTime())
if err != nil {
return nil, fmt.Errorf("invalid endTime (RFC3339): %w", err)
}
if !st.Before(et) {
return nil, fmt.Errorf("startTime must be before endTime")
}
chunks := splitByWeeksUTC(st, et)
if len(chunks) == 0 {
return &app_cloudep_blockchain.OKResp{}, nil
}
for _, ch := range chunks {
timeBucket := ch
err := l.svcCtx.WorkerPools.Submit(func() {
_ = l.svcCtx.BinanceDataSource.UpsertKline(context.Background(), usecase.QueryKline{
Symbol: in.GetSymbol(),
Interval: in.GetInterval(),
StartTime: st.UnixNano(),
EndTime: et.UnixNano(),
StartTime: timeBucket.start.UnixNano(),
EndTime: timeBucket.end.UnixNano(),
})
return
}()
})
if err != nil {
return nil, err
}
}
return &app_cloudep_blockchain.OKResp{}, nil
}
// --------- 小工具按「週一起」切段UTC ---------
type timeRange struct {
start time.Time // 含
end time.Time // 不含
}
// 取得該時間所在「週一 00:00:00」(UTC)
func weekStartMondayUTC(t time.Time) time.Time {
t = t.UTC()
// Go: Sunday=0 ... Saturday=6我們要週一=起始
wd := int(t.Weekday())
if wd == 0 { // Sunday
wd = 7
}
// 將時間歸零到當天 00:00
d0 := time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.UTC)
// 回推到週一
return d0.AddDate(0, 0, -(wd - 1))
}
// 將 [start,end) 切成多個「週粒度」的區間
func splitByWeeksUTC(start, end time.Time) []timeRange {
var chunks []timeRange
if !start.Before(end) {
return chunks
}
cur := start.UTC()
end = end.UTC()
for cur.Before(end) {
// 下一個週一 00:00
nextWeekStart := weekStartMondayUTC(cur).AddDate(0, 0, 7)
chunkEnd := end
if nextWeekStart.Before(end) {
chunkEnd = nextWeekStart
}
chunks = append(chunks, timeRange{start: cur, end: chunkEnd})
cur = chunkEnd
}
return chunks
}

View File

@ -0,0 +1,53 @@
package blockchainservicelogic
import (
"blockchain/internal/domain/usecase"
"context"
"fmt"
"time"
"blockchain/gen_result/pb/code.30cm.net/digimon/app-cloudep-blockchain"
"blockchain/internal/svc"
"github.com/zeromicro/go-zero/core/logx"
)
type ListCandleDataLogic struct {
ctx context.Context
svcCtx *svc.ServiceContext
logx.Logger
}
func NewListCandleDataLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ListCandleDataLogic {
return &ListCandleDataLogic{
ctx: ctx,
svcCtx: svcCtx,
Logger: logx.WithContext(ctx),
}
}
func (l *ListCandleDataLogic) ListCandleData(in *app_cloudep_blockchain.HistoryReq) (*app_cloudep_blockchain.ListCandleDataResp, error) {
// 1) 先驗證輸入(避免把錯誤時間丟進背景 goroutine
st, err := time.Parse(time.RFC3339, in.GetStartTime())
if err != nil {
return nil, fmt.Errorf("invalid startTime (RFC3339): %w", err)
}
et, err := time.Parse(time.RFC3339, in.GetEndTime())
if err != nil {
return nil, fmt.Errorf("invalid endTime (RFC3339): %w", err)
}
kline, err := l.svcCtx.BinanceDataSource.ListKline(l.ctx, usecase.QueryKline{
Symbol: in.GetSymbol(),
Interval: in.GetInterval(),
StartTime: st.UnixMilli(),
EndTime: et.UnixMilli(),
})
if err != nil {
return nil, err
}
fmt.Println(kline)
return &app_cloudep_blockchain.ListCandleDataResp{}, nil
}

View File

@ -2,10 +2,8 @@ package blockchainservicelogic
import (
app_cloudep_blockchain "blockchain/gen_result/pb/code.30cm.net/digimon/app-cloudep-blockchain"
"blockchain/internal/domain/usecase"
"blockchain/internal/svc"
"context"
"fmt"
"github.com/zeromicro/go-zero/core/logx"
)
@ -26,15 +24,15 @@ func NewPingLogic(ctx context.Context, svcCtx *svc.ServiceContext) *PingLogic {
func (l *PingLogic) Ping(_ *app_cloudep_blockchain.NoneReq) (*app_cloudep_blockchain.OKResp, error) {
kline, err := l.svcCtx.BinanceDataSource.ListKline(l.ctx, usecase.QueryKline{
Symbol: "ETHUSDT",
Interval: "1d",
StartTime: 1502928000000,
EndTime: 1503964800000,
})
if err != nil {
return nil, err
}
//kline, err := l.svcCtx.BinanceDataSource.ListKline(l.ctx, usecase.QueryKline{
// Symbol: "ETHUSDT",
// Interval: "1d",
// StartTime: 1502928000000,
// EndTime: 1503964800000,
//})
//if err != nil {
// return nil, err
//}
return &app_cloudep_blockchain.OKResp{}, nil
}

View File

@ -207,6 +207,7 @@ func (repo *BinanceRepository) GetKline(ctx context.Context, param repository.Qu
qb.LtOrEqNamed("open_time", "end"),
).ToCql()
fmt.Println(stmt, param.Symbol, param.Interval, param.StartTime, param.EndTime)
var result []entity.Kline
err := repo.db.GetSession().Query(stmt, names).BindMap(qb.M{
"symbol": param.Symbol,

View File

@ -1,5 +1,5 @@
// Code generated by goctl. DO NOT EDIT.
// goctl 1.8.1
// goctl 1.8.5
// Source: blockchain.proto
package server
@ -34,6 +34,11 @@ func (s *BlockchainServiceServer) GetHistoryKlineData(ctx context.Context, in *a
return l.GetHistoryKlineData(in)
}
func (s *BlockchainServiceServer) ListCandleData(ctx context.Context, in *app_cloudep_blockchain.HistoryReq) (*app_cloudep_blockchain.ListCandleDataResp, error) {
l := blockchainservicelogic.NewListCandleDataLogic(ctx, s.svcCtx)
return l.ListCandleData(in)
}
// Ping is a health-check endpoint.
func (s *BlockchainServiceServer) Ping(ctx context.Context, in *app_cloudep_blockchain.NoneReq) (*app_cloudep_blockchain.OKResp, error) {
l := blockchainservicelogic.NewPingLogic(ctx, s.svcCtx)

View File

@ -6,6 +6,7 @@ import (
"blockchain/internal/domain/usecase"
repo "blockchain/internal/repository"
uc "blockchain/internal/usecase"
"github.com/panjf2000/ants/v2"
"github.com/zeromicro/go-zero/core/stores/redis"
)
@ -13,6 +14,8 @@ type ServiceContext struct {
Config config.Config
BinanceDataSource usecase.DataSourceUseCase
BinanceRepo repository.DataSourceRepository
WorkerPools *ants.Pool
}
func NewServiceContext(c config.Config) *ServiceContext {
@ -32,6 +35,11 @@ func NewServiceContext(c config.Config) *ServiceContext {
DB: cassandra,
KeySpace: c.Cassandra.Keyspace,
})
wp, err := ants.NewPool(1024)
if err != nil {
panic(err)
}
// InitBinanceKLineWebsocket()
return &ServiceContext{
Config: c,
@ -39,5 +47,6 @@ func NewServiceContext(c config.Config) *ServiceContext {
BinanceDataSource: uc.MustBinanceUseCase(uc.BinanceUseCaseParam{
BinanceRepo: binanceRepo,
}),
WorkerPools: wp,
}
}