carrydela/cryptoHermes
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 初始化 cryptoHermes 行情网关 v1 MVP + harness 工程文档

Browse Source 
Binance USDⓈ-M Futures 行情网关，给上游 Hermes 量化分析引擎提供单一聚合
接口 /v1/market/context（K 线 + funding + OI + 多空比 + taker volume）。
只读公共行情，不下单、不接私钥、不查账户。

## v1 实现范围（Milestone 1-5）

- Clean Architecture 4 层（controller/usecase/repo/entity），接口边界在
  internal/usecase/ports.go
- Binance Futures REST client（K 线 / ticker24h / funding / OI / 多空比
  / taker volume 共 9 个接口），全链路 string 价格避免 float64 精度问题
- TimescaleDB 5 张 hypertable（market_klines / funding_rates /
  open_interest / long_short_ratio / taker_buy_sell_volume），主键含
  时间维度，UpsertMany 幂等
- robfig/cron 定时采集（15m/1h/4h/1d/1w 多周期 K 线 + 衍生品 15 分钟
  落库），未收线 K 线 (close_time > now) 由 mapper/repo 双重过滤
- pkg/httpclient 统一限流（默认 20 req/s, burst 40）+ 重试，避免触发
  Binance 2400 weight/min IP 上限
- /v1/market/context 聚合接口：errgroup 并发拉 snapshot/funding/OI，DB
  K 线不足 200 根回源 Binance 异步补
- cmd/backfill CLI 支持指定 from/to 大段回填（Binance 历史 OI / 多空比
  官方只保留 30 天，必须自己存）
- Docker Compose + Makefile + golang-migrate，本地一键启

技术指标（support/resistance/Vegas/箱体）留待 v2，技术段返回空对象 +
warning 占位。

## Harness 工程文档

- AGENTS.md — AI agent 工作速查（10 个章节）
- ai/project-map.md — 仓库结构、扩展点、控制流
- ai/risk-guardrails.md — G1-G10 守卫规则（每条带可机械验证命令）
- ai/adr/0001-architecture-foundations.md — 9 条架构基础决策
- ai/task-templates.md — 6 种任务契约模板
- ai/harness-health.md — 当前 harness 健康度评估

3 个 grep 守卫已验证通过：controller / usecase 无具体实现依赖，全项目
无私钥/签名字段。
This commit is contained in:
dela
2026-05-24 17:20:51 +08:00
commit cc7f5a4f32
58 changed files with 5356 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
internal/app/app.go Normal file
View File

@@ -0,0 +1,121 @@
package app
import (
"context"
"fmt"
"log/slog"
"os"
"os/signal"
"syscall"
"time"
"github.com/gofiber/fiber/v2"
"cryptoHermes/config"
"cryptoHermes/internal/controller/restapi"
"cryptoHermes/internal/repo/persistent/postgres"
"cryptoHermes/internal/repo/webapi/binance"
"cryptoHermes/internal/usecase"
"cryptoHermes/internal/worker"
pgpkg "cryptoHermes/pkg/postgres"
)
func Run(cfg *config.Config, log *slog.Logger) error {
rootCtx, cancel := context.WithCancel(context.Background())
defer cancel()
pool, err := pgpkg.NewPool(rootCtx, pgpkg.Options{
DSN: cfg.Postgres.DSN,
MaxConns: cfg.Postgres.MaxConns,
MinConns: cfg.Postgres.MinConns,
})
if err != nil {
return fmt.Errorf("postgres: %w", err)
}
defer pool.Close()
binanceClient := binance.NewClient(binance.ClientOptions{
BaseURL: cfg.Binance.BaseURL,
Timeout: cfg.Binance.Timeout,
RetryCount: cfg.Binance.RetryCount,
RPS: cfg.Binance.RPS,
Burst: cfg.Binance.Burst,
})
klineRepo := postgres.NewKlineRepo(pool)
fundingRepo := postgres.NewFundingRepo(pool)
oiRepo := postgres.NewOpenInterestRepo(pool)
lsRepo := postgres.NewLongShortRatioRepo(pool)
takerRepo := postgres.NewTakerVolumeRepo(pool)
contextUC := usecase.NewMarketContextUsecase(
binanceClient,
binanceClient,
klineRepo,
fundingRepo,
oiRepo,
lsRepo,
log,
)
collector := worker.NewCollector(worker.Deps{
MarketData: binanceClient,
Derivatives: binanceClient,
KlineRepo: klineRepo,
FundingRepo: fundingRepo,
OIRepo: oiRepo,
LSRepo: lsRepo,
TakerRepo: takerRepo,
Symbols: cfg.Collector.Symbols,
Intervals: cfg.Collector.Intervals,
Limit: cfg.Collector.DefaultLimit,
Logger: log,
})
var sched *Scheduler
if cfg.Collector.Enabled {
sched = NewScheduler(collector, log)
sched.Start(rootCtx)
defer sched.Stop()
}
fiberApp := fiber.New(fiber.Config{
AppName: cfg.App.Name,
ReadTimeout: cfg.HTTP.ReadTimeout,
WriteTimeout: cfg.HTTP.WriteTimeout,
IdleTimeout: cfg.HTTP.IdleTimeout,
ErrorHandler: restapi.ErrorHandler,
})
restapi.Register(fiberApp, restapi.Deps{
Logger: log,
MarketContext: contextUC,
MarketData: binanceClient,
Derivatives: binanceClient,
KlineRepo: klineRepo,
FundingRepo: fundingRepo,
OIRepo: oiRepo,
LSRepo: lsRepo,
})
listenErr := make(chan error, 1)
go func() {
addr := fmt.Sprintf(":%d", cfg.App.Port)
log.Info("http_listening", "addr", addr)
listenErr <- fiberApp.Listen(addr)
}()
stop := make(chan os.Signal, 1)
signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
select {
case err := <-listenErr:
return err
case sig := <-stop:
log.Info("shutdown_signal", "signal", sig.String())
}
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
defer shutdownCancel()
return fiberApp.ShutdownWithContext(shutdownCtx)
}

58
internal/app/scheduler.go Normal file
View File

@@ -0,0 +1,58 @@
package app
import (
"context"
"log/slog"
"github.com/robfig/cron/v3"
"cryptoHermes/internal/worker"
)
type Scheduler struct {
cron *cron.Cron
collector *worker.Collector
log *slog.Logger
rootCtx context.Context
}
func NewScheduler(c *worker.Collector, log *slog.Logger) *Scheduler {
return &Scheduler{
cron: cron.New(),
collector: c,
log: log,
}
}
func (s *Scheduler) Start(ctx context.Context) {
s.rootCtx = ctx
s.add("*/15 * * * *", "klines_15m", func(c context.Context) { _ = s.collector.CollectKlines(c, "15m") })
s.add("0 * * * *", "klines_1h", func(c context.Context) { _ = s.collector.CollectKlines(c, "1h") })
s.add("0 */4 * * *", "klines_4h", func(c context.Context) { _ = s.collector.CollectKlines(c, "4h") })
s.add("5 0 * * *", "klines_1d", func(c context.Context) { _ = s.collector.CollectKlines(c, "1d") })
s.add("10 0 * * 1", "klines_1w", func(c context.Context) { _ = s.collector.CollectKlines(c, "1w") })
s.add("*/15 * * * *", "funding", func(c context.Context) { _ = s.collector.CollectFunding(c) })
s.add("*/15 * * * *", "oi", func(c context.Context) { _ = s.collector.CollectOpenInterest(c) })
s.add("*/15 * * * *", "ls", func(c context.Context) { _ = s.collector.CollectLongShortRatio(c) })
s.add("*/15 * * * *", "taker", func(c context.Context) { _ = s.collector.CollectTakerVolume(c) })
s.cron.Start()
s.log.Info("scheduler_started")
}
func (s *Scheduler) add(spec, name string, fn func(context.Context)) {
_, err := s.cron.AddFunc(spec, func() {
s.log.Info("cron_tick", "job", name)
fn(s.rootCtx)
})
if err != nil {
s.log.Error("cron_add_failed", "job", name, "err", err)
}
}
func (s *Scheduler) Stop() {
s.log.Info("scheduler_stopping")
<-s.cron.Stop().Done()
}

38
internal/controller/restapi/middleware.go Normal file
View File

@@ -0,0 +1,38 @@
package restapi
import (
"log/slog"
"time"
"github.com/gofiber/fiber/v2"
)
func RequestLogger(log *slog.Logger) fiber.Handler {
return func(c *fiber.Ctx) error {
start := time.Now()
err := c.Next()
log.Info("http_request",
"method", c.Method(),
"path", c.Path(),
"status", c.Response().StatusCode(),
"duration_ms", time.Since(start).Milliseconds(),
"ip", c.IP(),
)
return err
}
}
func ErrorHandler(c *fiber.Ctx, err error) error {
code := fiber.StatusInternalServerError
msg := "internal error"
if fe, ok := err.(*fiber.Error); ok {
code = fe.Code
msg = fe.Message
}
return c.Status(code).JSON(fiber.Map{
"error": fiber.Map{
"code": code,
"message": msg,
},
})
}

39
internal/controller/restapi/router.go Normal file
View File

@@ -0,0 +1,39 @@
package restapi
import (
"log/slog"
"github.com/gofiber/fiber/v2"
"github.com/gofiber/fiber/v2/middleware/recover"
v1 "cryptoHermes/internal/controller/restapi/v1"
"cryptoHermes/internal/usecase"
)
type Deps struct {
Logger *slog.Logger
MarketContext *usecase.MarketContextUsecase
MarketData usecase.MarketDataProvider
Derivatives usecase.DerivativesProvider
KlineRepo usecase.KlineRepository
FundingRepo usecase.FundingRepository
OIRepo usecase.OpenInterestRepository
LSRepo usecase.LongShortRatioRepository
}
func Register(app *fiber.App, deps Deps) {
app.Use(recover.New())
app.Use(RequestLogger(deps.Logger))
api := app.Group("/v1")
v1.RegisterHealth(api)
v1.RegisterMarket(api, v1.MarketDeps{
MarketContext: deps.MarketContext,
MarketData: deps.MarketData,
Derivatives: deps.Derivatives,
KlineRepo: deps.KlineRepo,
FundingRepo: deps.FundingRepo,
OIRepo: deps.OIRepo,
LSRepo: deps.LSRepo,
})
}

16
internal/controller/restapi/v1/health_routes.go Normal file
View File

@@ -0,0 +1,16 @@
package v1
import (
"time"
"github.com/gofiber/fiber/v2"
)
func RegisterHealth(r fiber.Router) {
r.Get("/health", func(c *fiber.Ctx) error {
return c.JSON(fiber.Map{
"status": "ok",
"time": time.Now().UnixMilli(),
})
})
}

118
internal/controller/restapi/v1/market_routes.go Normal file
View File

@@ -0,0 +1,118 @@
package v1
import (
"strconv"
"strings"
"github.com/gofiber/fiber/v2"
"cryptoHermes/internal/entity"
"cryptoHermes/internal/usecase"
)
type MarketDeps struct {
MarketContext *usecase.MarketContextUsecase
MarketData usecase.MarketDataProvider
Derivatives usecase.DerivativesProvider
KlineRepo usecase.KlineRepository
FundingRepo usecase.FundingRepository
OIRepo usecase.OpenInterestRepository
LSRepo usecase.LongShortRatioRepository
}
var allowedIntervals = map[string]bool{
"15m": true, "1h": true, "4h": true, "1d": true, "1w": true,
}
func RegisterMarket(r fiber.Router, d MarketDeps) {
g := r.Group("/market")
g.Get("/context", func(c *fiber.Ctx) error {
symbol := strings.ToUpper(c.Query("symbol"))
if symbol == "" {
return fiber.NewError(fiber.StatusBadRequest, "symbol is required")
}
out, err := d.MarketContext.Build(c.UserContext(), symbol)
if err != nil && out == nil {
return fiber.NewError(fiber.StatusBadRequest, err.Error())
}
return c.JSON(out)
})
g.Get("/klines", func(c *fiber.Ctx) error {
symbol := strings.ToUpper(c.Query("symbol"))
interval := c.Query("interval")
if symbol == "" || interval == "" {
return fiber.NewError(fiber.StatusBadRequest, "symbol and interval are required")
}
if !allowedIntervals[interval] {
return fiber.NewError(fiber.StatusBadRequest, "interval must be one of 15m/1h/4h/1d/1w")
}
limit := 300
if l := c.Query("limit"); l != "" {
if v, err := strconv.Atoi(l); err == nil && v > 0 && v <= 1000 {
limit = v
}
}
rows, err := d.KlineRepo.FindRecent(c.UserContext(), symbol, interval, limit)
if err != nil {
return fiber.NewError(fiber.StatusInternalServerError, err.Error())
}
return c.JSON(rows)
})
g.Get("/snapshot", func(c *fiber.Ctx) error {
symbol := strings.ToUpper(c.Query("symbol"))
if symbol == "" {
return fiber.NewError(fiber.StatusBadRequest, "symbol is required")
}
t, err := d.MarketData.GetTicker24h(c.UserContext(), symbol)
if err != nil {
return fiber.NewError(fiber.StatusBadGateway, err.Error())
}
return c.JSON(t)
})
g.Get("/derivatives", func(c *fiber.Ctx) error {
symbol := strings.ToUpper(c.Query("symbol"))
period := c.Query("period", "1h")
if symbol == "" {
return fiber.NewError(fiber.StatusBadRequest, "symbol is required")
}
fundCur, ferr := d.Derivatives.GetCurrentFunding(c.UserContext(), symbol)
oiCur, oerr := d.Derivatives.GetCurrentOpenInterest(c.UserContext(), symbol)
fundHist, _ := d.FundingRepo.FindRecent(c.UserContext(), symbol, 100)
oiHist, _ := d.OIRepo.FindRecent(c.UserContext(), symbol, period, 200)
globalLS, _ := d.LSRepo.FindRecent(c.UserContext(), symbol, period, entity.RatioTypeGlobalAccount, 200)
topLS, _ := d.LSRepo.FindRecent(c.UserContext(), symbol, period, entity.RatioTypeTopTraderPosition, 200)
resp := fiber.Map{
"symbol": symbol,
"funding": fiber.Map{
"current": fundCur,
"history": fundHist,
"error": errString(ferr),
},
"openInterest": fiber.Map{
"current": oiCur,
"history": oiHist,
"error": errString(oerr),
},
"longShortRatio": fiber.Map{
"global": globalLS,
"topTraderPosition": topLS,
},
"takerBuySellVolume": []any{},
}
return c.JSON(resp)
})
}
func errString(e error) string {
if e == nil {
return ""
}
return e.Error()
}

9
internal/entity/funding.go Normal file
View File

@@ -0,0 +1,9 @@
package entity
type FundingRate struct {
Source string `json:"source"`
Symbol string `json:"symbol"`
FundingTime int64 `json:"fundingTime"`
FundingRate string `json:"fundingRate"`
MarkPrice string `json:"markPrice,omitempty"`
}

19
internal/entity/kline.go Normal file
View File

@@ -0,0 +1,19 @@
package entity
type Kline struct {
Source string `json:"source"`
Symbol string `json:"symbol"`
Interval string `json:"interval"`
OpenTime int64 `json:"openTime"`
CloseTime int64 `json:"closeTime"`
Open string `json:"open"`
High string `json:"high"`
Low string `json:"low"`
Close string `json:"close"`
Volume string `json:"volume"`
QuoteVolume string `json:"quoteVolume"`
TradeCount int64 `json:"tradeCount"`
TakerBuyBaseVolume string `json:"takerBuyBaseVolume"`
TakerBuyQuoteVolume string `json:"takerBuyQuoteVolume"`
IsClosed bool `json:"-"`
}

18
internal/entity/long_short_ratio.go Normal file
View File

@@ -0,0 +1,18 @@
package entity
const (
RatioTypeGlobalAccount = "global_account"
RatioTypeTopTraderPosition = "top_trader_position"
RatioTypeTopTraderAccount = "top_trader_account"
)
type LongShortRatio struct {
Source string `json:"source"`
Symbol string `json:"symbol"`
Period string `json:"period"`
RatioType string `json:"ratioType"`
Timestamp int64 `json:"timestamp"`
LongShortRatio string `json:"longShortRatio"`
LongValue string `json:"longValue,omitempty"`
ShortValue string `json:"shortValue,omitempty"`
}

52
internal/entity/market_context.go Normal file
View File

@@ -0,0 +1,52 @@
package entity
type MarketContext struct {
Symbol string `json:"symbol"`
GeneratedAt int64 `json:"generatedAt"`
Snapshot *Ticker24h `json:"snapshot"`
Klines map[string][]Kline `json:"klines"`
Derivatives DerivativesBundle `json:"derivatives"`
Technical TechnicalStructure `json:"technical"`
DataQuality DataQuality `json:"dataQuality"`
}
type DerivativesBundle struct {
Funding FundingBundle `json:"funding"`
OpenInterest OpenInterestBundle `json:"openInterest"`
LongShortRatio LongShortBundle `json:"longShortRatio"`
TakerBuySellVolume []TakerBuySellVolume `json:"takerBuySellVolume"`
}
type FundingBundle struct {
Current *FundingRate `json:"current"`
History []FundingRate `json:"history"`
}
type OpenInterestBundle struct {
Current *OpenInterest `json:"current"`
History []OpenInterest `json:"history"`
}
type LongShortBundle struct {
Global []LongShortRatio `json:"global"`
TopTraderPosition []LongShortRatio `json:"topTraderPosition"`
}
type TechnicalStructure struct {
Support []TechnicalLevel `json:"support"`
Resistance []TechnicalLevel `json:"resistance"`
RangeHigh *string `json:"rangeHigh"`
RangeLow *string `json:"rangeLow"`
LongShortLine *string `json:"longShortLine"`
}
type TechnicalLevel struct {
Price string `json:"price"`
Strength string `json:"strength,omitempty"`
Source string `json:"source,omitempty"`
}
type DataQuality struct {
Source string `json:"source"`
Warnings []string `json:"warnings"`
}

10
internal/entity/open_interest.go Normal file
View File

@@ -0,0 +1,10 @@
package entity
type OpenInterest struct {
Source string `json:"source"`
Symbol string `json:"symbol"`
Period string `json:"period"`
Timestamp int64 `json:"timestamp"`
OpenInterest string `json:"openInterest"`
OpenInterestValue string `json:"openInterestValue,omitempty"`
}

11
internal/entity/taker_volume.go Normal file
View File

@@ -0,0 +1,11 @@
package entity
type TakerBuySellVolume struct {
Source string `json:"source"`
Symbol string `json:"symbol"`
Period string `json:"period"`
Timestamp int64 `json:"timestamp"`
BuySellRatio string `json:"buySellRatio,omitempty"`
BuyVolume string `json:"buyVolume,omitempty"`
SellVolume string `json:"sellVolume,omitempty"`
}

14
internal/entity/ticker.go Normal file
View File

@@ -0,0 +1,14 @@
package entity
type Ticker24h struct {
Symbol string `json:"symbol"`
LastPrice string `json:"lastPrice"`
PriceChange string `json:"priceChange"`
PriceChangePercent string `json:"priceChangePercent"`
HighPrice string `json:"highPrice"`
LowPrice string `json:"lowPrice"`
Volume string `json:"volume"`
QuoteVolume string `json:"quoteVolume"`
OpenTime int64 `json:"openTime"`
CloseTime int64 `json:"closeTime"`
}

86
internal/repo/persistent/postgres/funding_repo.go Normal file
View File

@@ -0,0 +1,86 @@
package postgres
import (
"context"
"database/sql"
"fmt"
"github.com/jackc/pgx/v5/pgxpool"
"cryptoHermes/internal/entity"
)
type FundingRepo struct {
pool *pgxpool.Pool
}
func NewFundingRepo(pool *pgxpool.Pool) *FundingRepo {
return &FundingRepo{pool: pool}
}
const fundingUpsertSQL = `
INSERT INTO funding_rates (source, symbol, funding_time, funding_rate, mark_price)
VALUES ($1, $2, $3, $4, NULLIF($5, '')::NUMERIC)
ON CONFLICT (source, symbol, funding_time) DO UPDATE SET
funding_rate = EXCLUDED.funding_rate,
mark_price = EXCLUDED.mark_price
`
func (r *FundingRepo) UpsertMany(ctx context.Context, items []entity.FundingRate) error {
if len(items) == 0 {
return nil
}
tx, err := r.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
for _, f := range items {
if f.FundingRate == "" || f.FundingTime == 0 {
continue
}
if _, err := tx.Exec(ctx, fundingUpsertSQL,
f.Source, f.Symbol, f.FundingTime, f.FundingRate, f.MarkPrice,
); err != nil {
return fmt.Errorf("upsert funding %s@%d: %w", f.Symbol, f.FundingTime, err)
}
}
return tx.Commit(ctx)
}
const fundingFindSQL = `
SELECT source, symbol, funding_time, funding_rate::text, COALESCE(mark_price::text, '')
FROM funding_rates
WHERE symbol = $1
ORDER BY funding_time DESC
LIMIT $2
`
func (r *FundingRepo) FindRecent(ctx context.Context, symbol string, limit int) ([]entity.FundingRate, error) {
if limit <= 0 {
limit = 100
}
rows, err := r.pool.Query(ctx, fundingFindSQL, symbol, limit)
if err != nil {
return nil, err
}
defer rows.Close()
out := make([]entity.FundingRate, 0, limit)
for rows.Next() {
var f entity.FundingRate
var mark sql.NullString
if err := rows.Scan(&f.Source, &f.Symbol, &f.FundingTime, &f.FundingRate, &mark); err != nil {
return nil, err
}
if mark.Valid {
f.MarkPrice = mark.String
}
out = append(out, f)
}
for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
out[i], out[j] = out[j], out[i]
}
return out, rows.Err()
}

114
internal/repo/persistent/postgres/kline_repo.go Normal file
View File

@@ -0,0 +1,114 @@
package postgres
import (
"context"
"fmt"
"github.com/jackc/pgx/v5/pgxpool"
"cryptoHermes/internal/entity"
)
type KlineRepo struct {
pool *pgxpool.Pool
}
func NewKlineRepo(pool *pgxpool.Pool) *KlineRepo {
return &KlineRepo{pool: pool}
}
const klineUpsertSQL = `
INSERT INTO market_klines (
source, symbol, interval, open_time, close_time,
open, high, low, close,
volume, quote_volume,
trade_count, taker_buy_base_volume, taker_buy_quote_volume,
updated_at
) VALUES (
$1, $2, $3, $4, $5,
$6, $7, $8, $9,
$10, $11,
$12, $13, $14,
now()
)
ON CONFLICT (source, symbol, interval, open_time) DO UPDATE SET
close_time = EXCLUDED.close_time,
open = EXCLUDED.open,
high = EXCLUDED.high,
low = EXCLUDED.low,
close = EXCLUDED.close,
volume = EXCLUDED.volume,
quote_volume = EXCLUDED.quote_volume,
trade_count = EXCLUDED.trade_count,
taker_buy_base_volume = EXCLUDED.taker_buy_base_volume,
taker_buy_quote_volume = EXCLUDED.taker_buy_quote_volume,
updated_at = now()
`
func (r *KlineRepo) UpsertMany(ctx context.Context, items []entity.Kline) error {
if len(items) == 0 {
return nil
}
tx, err := r.pool.Begin(ctx)
if err != nil {
return fmt.Errorf("begin: %w", err)
}
defer tx.Rollback(ctx)
for _, k := range items {
if !k.IsClosed {
continue
}
_, err := tx.Exec(ctx, klineUpsertSQL,
k.Source, k.Symbol, k.Interval, k.OpenTime, k.CloseTime,
k.Open, k.High, k.Low, k.Close,
k.Volume, k.QuoteVolume,
k.TradeCount, k.TakerBuyBaseVolume, k.TakerBuyQuoteVolume,
)
if err != nil {
return fmt.Errorf("upsert kline %s/%s@%d: %w", k.Symbol, k.Interval, k.OpenTime, err)
}
}
return tx.Commit(ctx)
}
const klineFindSQL = `
SELECT source, symbol, interval, open_time, close_time,
open::text, high::text, low::text, close::text,
volume::text, quote_volume::text,
trade_count, taker_buy_base_volume::text, taker_buy_quote_volume::text
FROM market_klines
WHERE symbol = $1 AND interval = $2
ORDER BY open_time DESC
LIMIT $3
`
func (r *KlineRepo) FindRecent(ctx context.Context, symbol, interval string, limit int) ([]entity.Kline, error) {
if limit <= 0 {
limit = 300
}
rows, err := r.pool.Query(ctx, klineFindSQL, symbol, interval, limit)
if err != nil {
return nil, fmt.Errorf("query: %w", err)
}
defer rows.Close()
out := make([]entity.Kline, 0, limit)
for rows.Next() {
var k entity.Kline
if err := rows.Scan(
&k.Source, &k.Symbol, &k.Interval, &k.OpenTime, &k.CloseTime,
&k.Open, &k.High, &k.Low, &k.Close,
&k.Volume, &k.QuoteVolume,
&k.TradeCount, &k.TakerBuyBaseVolume, &k.TakerBuyQuoteVolume,
); err != nil {
return nil, fmt.Errorf("scan: %w", err)
}
k.IsClosed = true
out = append(out, k)
}
for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
out[i], out[j] = out[j], out[i]
}
return out, rows.Err()
}

94
internal/repo/persistent/postgres/long_short_ratio_repo.go Normal file
View File

@@ -0,0 +1,94 @@
package postgres
import (
"context"
"fmt"
"github.com/jackc/pgx/v5/pgxpool"
"cryptoHermes/internal/entity"
)
type LongShortRatioRepo struct {
pool *pgxpool.Pool
}
func NewLongShortRatioRepo(pool *pgxpool.Pool) *LongShortRatioRepo {
return &LongShortRatioRepo{pool: pool}
}
const lsUpsertSQL = `
INSERT INTO long_short_ratio (
source, symbol, period, ratio_type, timestamp,
long_short_ratio, long_value, short_value
) VALUES (
$1, $2, $3, $4, $5,
$6,
NULLIF($7, '')::NUMERIC,
NULLIF($8, '')::NUMERIC
)
ON CONFLICT (source, symbol, period, ratio_type, timestamp) DO UPDATE SET
long_short_ratio = EXCLUDED.long_short_ratio,
long_value = EXCLUDED.long_value,
short_value = EXCLUDED.short_value
`
func (r *LongShortRatioRepo) UpsertMany(ctx context.Context, items []entity.LongShortRatio) error {
if len(items) == 0 {
return nil
}
tx, err := r.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
for _, ls := range items {
if ls.LongShortRatio == "" || ls.Timestamp == 0 {
continue
}
if _, err := tx.Exec(ctx, lsUpsertSQL,
ls.Source, ls.Symbol, ls.Period, ls.RatioType, ls.Timestamp,
ls.LongShortRatio, ls.LongValue, ls.ShortValue,
); err != nil {
return fmt.Errorf("upsert ls %s/%s/%s@%d: %w", ls.Symbol, ls.Period, ls.RatioType, ls.Timestamp, err)
}
}
return tx.Commit(ctx)
}
const lsFindSQL = `
SELECT source, symbol, period, ratio_type, timestamp,
long_short_ratio::text,
COALESCE(long_value::text, ''),
COALESCE(short_value::text, '')
FROM long_short_ratio
WHERE symbol = $1 AND period = $2 AND ratio_type = $3
ORDER BY timestamp DESC
LIMIT $4
`
func (r *LongShortRatioRepo) FindRecent(ctx context.Context, symbol, period, ratioType string, limit int) ([]entity.LongShortRatio, error) {
if limit <= 0 {
limit = 100
}
rows, err := r.pool.Query(ctx, lsFindSQL, symbol, period, ratioType, limit)
if err != nil {
return nil, err
}
defer rows.Close()
out := make([]entity.LongShortRatio, 0, limit)
for rows.Next() {
var ls entity.LongShortRatio
if err := rows.Scan(&ls.Source, &ls.Symbol, &ls.Period, &ls.RatioType, &ls.Timestamp,
&ls.LongShortRatio, &ls.LongValue, &ls.ShortValue); err != nil {
return nil, err
}
out = append(out, ls)
}
for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
out[i], out[j] = out[j], out[i]
}
return out, rows.Err()
}

82
internal/repo/persistent/postgres/open_interest_repo.go Normal file
View File

@@ -0,0 +1,82 @@
package postgres
import (
"context"
"fmt"
"github.com/jackc/pgx/v5/pgxpool"
"cryptoHermes/internal/entity"
)
type OpenInterestRepo struct {
pool *pgxpool.Pool
}
func NewOpenInterestRepo(pool *pgxpool.Pool) *OpenInterestRepo {
return &OpenInterestRepo{pool: pool}
}
const oiUpsertSQL = `
INSERT INTO open_interest (source, symbol, period, timestamp, open_interest, open_interest_value)
VALUES ($1, $2, $3, $4, $5, NULLIF($6, '')::NUMERIC)
ON CONFLICT (source, symbol, period, timestamp) DO UPDATE SET
open_interest = EXCLUDED.open_interest,
open_interest_value = EXCLUDED.open_interest_value
`
func (r *OpenInterestRepo) UpsertMany(ctx context.Context, items []entity.OpenInterest) error {
if len(items) == 0 {
return nil
}
tx, err := r.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
for _, o := range items {
if o.OpenInterest == "" || o.Timestamp == 0 {
continue
}
if _, err := tx.Exec(ctx, oiUpsertSQL,
o.Source, o.Symbol, o.Period, o.Timestamp, o.OpenInterest, o.OpenInterestValue,
); err != nil {
return fmt.Errorf("upsert oi %s/%s@%d: %w", o.Symbol, o.Period, o.Timestamp, err)
}
}
return tx.Commit(ctx)
}
const oiFindSQL = `
SELECT source, symbol, period, timestamp,
open_interest::text, COALESCE(open_interest_value::text, '')
FROM open_interest
WHERE symbol = $1 AND period = $2
ORDER BY timestamp DESC
LIMIT $3
`
func (r *OpenInterestRepo) FindRecent(ctx context.Context, symbol, period string, limit int) ([]entity.OpenInterest, error) {
if limit <= 0 {
limit = 100
}
rows, err := r.pool.Query(ctx, oiFindSQL, symbol, period, limit)
if err != nil {
return nil, err
}
defer rows.Close()
out := make([]entity.OpenInterest, 0, limit)
for rows.Next() {
var o entity.OpenInterest
if err := rows.Scan(&o.Source, &o.Symbol, &o.Period, &o.Timestamp, &o.OpenInterest, &o.OpenInterestValue); err != nil {
return nil, err
}
out = append(out, o)
}
for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
out[i], out[j] = out[j], out[i]
}
return out, rows.Err()
}

90
internal/repo/persistent/postgres/taker_volume_repo.go Normal file
View File

@@ -0,0 +1,90 @@
package postgres
import (
"context"
"fmt"
"github.com/jackc/pgx/v5/pgxpool"
"cryptoHermes/internal/entity"
)
type TakerVolumeRepo struct {
pool *pgxpool.Pool
}
func NewTakerVolumeRepo(pool *pgxpool.Pool) *TakerVolumeRepo {
return &TakerVolumeRepo{pool: pool}
}
const takerUpsertSQL = `
INSERT INTO taker_buy_sell_volume (source, symbol, period, timestamp, buy_sell_ratio, buy_volume, sell_volume)
VALUES ($1, $2, $3, $4,
NULLIF($5, '')::NUMERIC,
NULLIF($6, '')::NUMERIC,
NULLIF($7, '')::NUMERIC)
ON CONFLICT (source, symbol, period, timestamp) DO UPDATE SET
buy_sell_ratio = EXCLUDED.buy_sell_ratio,
buy_volume = EXCLUDED.buy_volume,
sell_volume = EXCLUDED.sell_volume
`
func (r *TakerVolumeRepo) UpsertMany(ctx context.Context, items []entity.TakerBuySellVolume) error {
if len(items) == 0 {
return nil
}
tx, err := r.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
for _, t := range items {
if t.Timestamp == 0 {
continue
}
if _, err := tx.Exec(ctx, takerUpsertSQL,
t.Source, t.Symbol, t.Period, t.Timestamp,
t.BuySellRatio, t.BuyVolume, t.SellVolume,
); err != nil {
return fmt.Errorf("upsert taker %s/%s@%d: %w", t.Symbol, t.Period, t.Timestamp, err)
}
}
return tx.Commit(ctx)
}
const takerFindSQL = `
SELECT source, symbol, period, timestamp,
COALESCE(buy_sell_ratio::text, ''),
COALESCE(buy_volume::text, ''),
COALESCE(sell_volume::text, '')
FROM taker_buy_sell_volume
WHERE symbol = $1 AND period = $2
ORDER BY timestamp DESC
LIMIT $3
`
func (r *TakerVolumeRepo) FindRecent(ctx context.Context, symbol, period string, limit int) ([]entity.TakerBuySellVolume, error) {
if limit <= 0 {
limit = 100
}
rows, err := r.pool.Query(ctx, takerFindSQL, symbol, period, limit)
if err != nil {
return nil, err
}
defer rows.Close()
out := make([]entity.TakerBuySellVolume, 0, limit)
for rows.Next() {
var t entity.TakerBuySellVolume
if err := rows.Scan(&t.Source, &t.Symbol, &t.Period, &t.Timestamp,
&t.BuySellRatio, &t.BuyVolume, &t.SellVolume); err != nil {
return nil, err
}
out = append(out, t)
}
for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
out[i], out[j] = out[j], out[i]
}
return out, rows.Err()
}

72
internal/repo/webapi/binance/client.go Normal file
View File

@@ -0,0 +1,72 @@
package binance
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/url"
"time"
"cryptoHermes/pkg/httpclient"
)
const sourceName = "binance"
type Client struct {
http *httpclient.Client
source string
}
type ClientOptions struct {
BaseURL string
Timeout time.Duration
RetryCount int
RPS float64
Burst int
}
func NewClient(opts ClientOptions) *Client {
return &Client{
http: httpclient.New(httpclient.Options{
BaseURL: opts.BaseURL,
Timeout: opts.Timeout,
RetryCount: opts.RetryCount,
RPS: opts.RPS,
Burst: opts.Burst,
}),
source: sourceName,
}
}
type ExternalAPIError struct {
Source string
Path string
StatusCode int
Message string
}
func (e *ExternalAPIError) Error() string {
return fmt.Sprintf("%s %s status=%d: %s", e.Source, e.Path, e.StatusCode, e.Message)
}
func (c *Client) get(ctx context.Context, path string, q url.Values, out any) error {
status, body, _, err := c.http.Do(ctx, httpclient.Request{
Method: http.MethodGet,
Path: path,
Query: q,
})
if err != nil {
return &ExternalAPIError{Source: c.source, Path: path, StatusCode: 0, Message: err.Error()}
}
if status >= 400 {
return &ExternalAPIError{Source: c.source, Path: path, StatusCode: status, Message: string(body)}
}
if out == nil {
return nil
}
if err := json.Unmarshal(body, out); err != nil {
return &ExternalAPIError{Source: c.source, Path: path, StatusCode: status, Message: "decode: " + err.Error()}
}
return nil
}

116
internal/repo/webapi/binance/derivatives.go Normal file
View File

@@ -0,0 +1,116 @@
package binance
import (
"context"
"net/url"
"strconv"
"cryptoHermes/internal/entity"
)
func (c *Client) GetCurrentFunding(ctx context.Context, symbol string) (*entity.FundingRate, error) {
q := url.Values{}
q.Set("symbol", symbol)
var dto premiumIndexDTO
if err := c.get(ctx, "/fapi/v1/premiumIndex", q, &dto); err != nil {
return nil, err
}
return mapPremiumIndex(&dto), nil
}
func (c *Client) GetFundingHistory(ctx context.Context, symbol string, limit int) ([]entity.FundingRate, error) {
if limit <= 0 {
limit = 100
}
if limit > 1000 {
limit = 1000
}
q := url.Values{}
q.Set("symbol", symbol)
q.Set("limit", strconv.Itoa(limit))
var dto []fundingRateDTO
if err := c.get(ctx, "/fapi/v1/fundingRate", q, &dto); err != nil {
return nil, err
}
return mapFundingHistory(dto), nil
}
func (c *Client) GetCurrentOpenInterest(ctx context.Context, symbol string) (*entity.OpenInterest, error) {
q := url.Values{}
q.Set("symbol", symbol)
var dto openInterestDTO
if err := c.get(ctx, "/fapi/v1/openInterest", q, &dto); err != nil {
return nil, err
}
return mapOpenInterest(&dto, "current"), nil
}
func (c *Client) GetOpenInterestHistory(ctx context.Context, symbol, period string, limit int) ([]entity.OpenInterest, error) {
if limit <= 0 {
limit = 30
}
if limit > 500 {
limit = 500
}
q := url.Values{}
q.Set("symbol", symbol)
q.Set("period", period)
q.Set("limit", strconv.Itoa(limit))
var dto []openInterestHistDTO
if err := c.get(ctx, "/futures/data/openInterestHist", q, &dto); err != nil {
return nil, err
}
return mapOpenInterestHistory(dto, period), nil
}
func (c *Client) GetGlobalLongShortRatio(ctx context.Context, symbol, period string, limit int) ([]entity.LongShortRatio, error) {
dto, err := c.fetchLongShort(ctx, "/futures/data/globalLongShortAccountRatio", symbol, period, limit)
if err != nil {
return nil, err
}
return mapLongShort(dto, period, entity.RatioTypeGlobalAccount), nil
}
func (c *Client) GetTopTraderPositionRatio(ctx context.Context, symbol, period string, limit int) ([]entity.LongShortRatio, error) {
dto, err := c.fetchLongShort(ctx, "/futures/data/topLongShortPositionRatio", symbol, period, limit)
if err != nil {
return nil, err
}
return mapLongShort(dto, period, entity.RatioTypeTopTraderPosition), nil
}
func (c *Client) fetchLongShort(ctx context.Context, path, symbol, period string, limit int) ([]longShortRatioDTO, error) {
if limit <= 0 {
limit = 30
}
if limit > 500 {
limit = 500
}
q := url.Values{}
q.Set("symbol", symbol)
q.Set("period", period)
q.Set("limit", strconv.Itoa(limit))
var dto []longShortRatioDTO
if err := c.get(ctx, path, q, &dto); err != nil {
return nil, err
}
return dto, nil
}
func (c *Client) GetTakerBuySellVolume(ctx context.Context, symbol, period string, limit int) ([]entity.TakerBuySellVolume, error) {
if limit <= 0 {
limit = 30
}
if limit > 500 {
limit = 500
}
q := url.Values{}
q.Set("symbol", symbol)
q.Set("period", period)
q.Set("limit", strconv.Itoa(limit))
var dto []takerVolumeDTO
if err := c.get(ctx, "/futures/data/takerlongshortRatio", q, &dto); err != nil {
return nil, err
}
return mapTakerVolume(dto, symbol, period), nil
}

60
internal/repo/webapi/binance/dto.go Normal file
View File

@@ -0,0 +1,60 @@
package binance
type tickerDTO struct {
Symbol string `json:"symbol"`
LastPrice string `json:"lastPrice"`
PriceChange string `json:"priceChange"`
PriceChangePercent string `json:"priceChangePercent"`
HighPrice string `json:"highPrice"`
LowPrice string `json:"lowPrice"`
Volume string `json:"volume"`
QuoteVolume string `json:"quoteVolume"`
OpenTime int64 `json:"openTime"`
CloseTime int64 `json:"closeTime"`
}
type premiumIndexDTO struct {
Symbol string `json:"symbol"`
MarkPrice string `json:"markPrice"`
IndexPrice string `json:"indexPrice"`
LastFundingRate string `json:"lastFundingRate"`
NextFundingTime int64 `json:"nextFundingTime"`
Time int64 `json:"time"`
}
type fundingRateDTO struct {
Symbol string `json:"symbol"`
FundingTime int64 `json:"fundingTime"`
FundingRate string `json:"fundingRate"`
MarkPrice string `json:"markPrice"`
}
type openInterestDTO struct {
OpenInterest string `json:"openInterest"`
Symbol string `json:"symbol"`
Time int64 `json:"time"`
}
type openInterestHistDTO struct {
Symbol string `json:"symbol"`
SumOpenInterest string `json:"sumOpenInterest"`
SumOpenInterestValue string `json:"sumOpenInterestValue"`
Timestamp int64 `json:"timestamp"`
}
type longShortRatioDTO struct {
Symbol string `json:"symbol"`
LongShortRatio string `json:"longShortRatio"`
LongAccount string `json:"longAccount"`
ShortAccount string `json:"shortAccount"`
LongPosition string `json:"longPosition"`
ShortPosition string `json:"shortPosition"`
Timestamp int64 `json:"timestamp"`
}
type takerVolumeDTO struct {
BuySellRatio string `json:"buySellRatio"`
BuyVol string `json:"buyVol"`
SellVol string `json:"sellVol"`
Timestamp int64 `json:"timestamp"`
}

170
internal/repo/webapi/binance/mapper.go Normal file
View File

@@ -0,0 +1,170 @@
package binance
import (
"fmt"
"time"
"cryptoHermes/internal/entity"
)
func parseKlineRow(symbol, interval string, row []any) (entity.Kline, error) {
if len(row) < 12 {
return entity.Kline{}, fmt.Errorf("binance kline row has %d fields, expected >=12", len(row))
}
openTime, err := toInt64(row[0])
if err != nil {
return entity.Kline{}, fmt.Errorf("openTime: %w", err)
}
closeTime, err := toInt64(row[6])
if err != nil {
return entity.Kline{}, fmt.Errorf("closeTime: %w", err)
}
tradeCount, err := toInt64(row[8])
if err != nil {
return entity.Kline{}, fmt.Errorf("tradeCount: %w", err)
}
open, _ := row[1].(string)
high, _ := row[2].(string)
low, _ := row[3].(string)
closeP, _ := row[4].(string)
volume, _ := row[5].(string)
quoteVol, _ := row[7].(string)
takerBase, _ := row[9].(string)
takerQuote, _ := row[10].(string)
nowMs := time.Now().UnixMilli()
return entity.Kline{
Source: sourceName,
Symbol: symbol,
Interval: interval,
OpenTime: openTime,
CloseTime: closeTime,
Open: open,
High: high,
Low: low,
Close: closeP,
Volume: volume,
QuoteVolume: quoteVol,
TradeCount: tradeCount,
TakerBuyBaseVolume: takerBase,
TakerBuyQuoteVolume: takerQuote,
IsClosed: closeTime < nowMs,
}, nil
}
func toInt64(v any) (int64, error) {
switch x := v.(type) {
case float64:
return int64(x), nil
case int64:
return x, nil
case int:
return int64(x), nil
default:
return 0, fmt.Errorf("unexpected type %T", v)
}
}
func mapTicker(d *tickerDTO) *entity.Ticker24h {
return &entity.Ticker24h{
Symbol: d.Symbol,
LastPrice: d.LastPrice,
PriceChange: d.PriceChange,
PriceChangePercent: d.PriceChangePercent,
HighPrice: d.HighPrice,
LowPrice: d.LowPrice,
Volume: d.Volume,
QuoteVolume: d.QuoteVolume,
OpenTime: d.OpenTime,
CloseTime: d.CloseTime,
}
}
func mapPremiumIndex(d *premiumIndexDTO) *entity.FundingRate {
return &entity.FundingRate{
Source: sourceName,
Symbol: d.Symbol,
FundingTime: d.NextFundingTime,
FundingRate: d.LastFundingRate,
MarkPrice: d.MarkPrice,
}
}
func mapFundingHistory(d []fundingRateDTO) []entity.FundingRate {
out := make([]entity.FundingRate, 0, len(d))
for _, x := range d {
out = append(out, entity.FundingRate{
Source: sourceName,
Symbol: x.Symbol,
FundingTime: x.FundingTime,
FundingRate: x.FundingRate,
MarkPrice: x.MarkPrice,
})
}
return out
}
func mapOpenInterest(d *openInterestDTO, period string) *entity.OpenInterest {
return &entity.OpenInterest{
Source: sourceName,
Symbol: d.Symbol,
Period: period,
Timestamp: d.Time,
OpenInterest: d.OpenInterest,
}
}
func mapOpenInterestHistory(d []openInterestHistDTO, period string) []entity.OpenInterest {
out := make([]entity.OpenInterest, 0, len(d))
for _, x := range d {
out = append(out, entity.OpenInterest{
Source: sourceName,
Symbol: x.Symbol,
Period: period,
Timestamp: x.Timestamp,
OpenInterest: x.SumOpenInterest,
OpenInterestValue: x.SumOpenInterestValue,
})
}
return out
}
func mapLongShort(d []longShortRatioDTO, period, ratioType string) []entity.LongShortRatio {
out := make([]entity.LongShortRatio, 0, len(d))
for _, x := range d {
longVal := x.LongAccount
shortVal := x.ShortAccount
if ratioType == entity.RatioTypeTopTraderPosition {
longVal = x.LongPosition
shortVal = x.ShortPosition
}
out = append(out, entity.LongShortRatio{
Source: sourceName,
Symbol: x.Symbol,
Period: period,
RatioType: ratioType,
Timestamp: x.Timestamp,
LongShortRatio: x.LongShortRatio,
LongValue: longVal,
ShortValue: shortVal,
})
}
return out
}
func mapTakerVolume(d []takerVolumeDTO, symbol, period string) []entity.TakerBuySellVolume {
out := make([]entity.TakerBuySellVolume, 0, len(d))
for _, x := range d {
out = append(out, entity.TakerBuySellVolume{
Source: sourceName,
Symbol: symbol,
Period: period,
Timestamp: x.Timestamp,
BuySellRatio: x.BuySellRatio,
BuyVolume: x.BuyVol,
SellVolume: x.SellVol,
})
}
return out
}

57
internal/repo/webapi/binance/market.go Normal file
View File

@@ -0,0 +1,57 @@
package binance
import (
"context"
"net/url"
"strconv"
"cryptoHermes/internal/entity"
)
func (c *Client) GetKlines(ctx context.Context, symbol, interval string, limit int) ([]entity.Kline, error) {
return c.GetKlinesRange(ctx, symbol, interval, 0, 0, limit)
}
func (c *Client) GetKlinesRange(ctx context.Context, symbol, interval string, startMs, endMs int64, limit int) ([]entity.Kline, error) {
if limit <= 0 {
limit = 500
}
if limit > 1500 {
limit = 1500
}
q := url.Values{}
q.Set("symbol", symbol)
q.Set("interval", interval)
q.Set("limit", strconv.Itoa(limit))
if startMs > 0 {
q.Set("startTime", strconv.FormatInt(startMs, 10))
}
if endMs > 0 {
q.Set("endTime", strconv.FormatInt(endMs, 10))
}
var raw [][]any
if err := c.get(ctx, "/fapi/v1/klines", q, &raw); err != nil {
return nil, err
}
out := make([]entity.Kline, 0, len(raw))
for _, row := range raw {
k, err := parseKlineRow(symbol, interval, row)
if err != nil {
return nil, err
}
out = append(out, k)
}
return out, nil
}
func (c *Client) GetTicker24h(ctx context.Context, symbol string) (*entity.Ticker24h, error) {
q := url.Values{}
q.Set("symbol", symbol)
var dto tickerDTO
if err := c.get(ctx, "/fapi/v1/ticker/24hr", q, &dto); err != nil {
return nil, err
}
return mapTicker(&dto), nil
}

18
internal/usecase/market_collector.go Normal file
View File

@@ -0,0 +1,18 @@
package usecase
import (
"context"
"log/slog"
)
type MarketCollectorUsecase struct {
log *slog.Logger
}
func NewMarketCollectorUsecase(log *slog.Logger) *MarketCollectorUsecase {
return &MarketCollectorUsecase{log: log}
}
func (u *MarketCollectorUsecase) Tick(ctx context.Context) error {
return nil
}

204
internal/usecase/market_context.go Normal file
View File

@@ -0,0 +1,204 @@
package usecase
import (
"context"
"errors"
"fmt"
"log/slog"
"sync"
"time"
"golang.org/x/sync/errgroup"
"cryptoHermes/internal/entity"
)
var supportedSymbols = map[string]bool{
"BTCUSDT": true,
"ETHUSDT": true,
}
var supportedIntervals = []string{"15m", "1h", "4h", "1d", "1w"}
const (
klineWindowSize = 300
klineMinForOK = 200
derivativePeriod = "1h"
fundingHistoryLen = 100
oiHistoryLen = 200
longShortLen = 200
takerHistoryLen = 200
)
type MarketContextUsecase struct {
marketData MarketDataProvider
derivatives DerivativesProvider
klineRepo KlineRepository
fundingRepo FundingRepository
oiRepo OpenInterestRepository
lsRepo LongShortRatioRepository
log *slog.Logger
}
func NewMarketContextUsecase(
marketData MarketDataProvider,
derivatives DerivativesProvider,
klineRepo KlineRepository,
fundingRepo FundingRepository,
oiRepo OpenInterestRepository,
lsRepo LongShortRatioRepository,
log *slog.Logger,
) *MarketContextUsecase {
return &MarketContextUsecase{
marketData: marketData,
derivatives: derivatives,
klineRepo: klineRepo,
fundingRepo: fundingRepo,
oiRepo: oiRepo,
lsRepo: lsRepo,
log: log,
}
}
func (u *MarketContextUsecase) Build(ctx context.Context, symbol string) (*entity.MarketContext, error) {
if !supportedSymbols[symbol] {
return nil, fmt.Errorf("unsupported symbol: %s", symbol)
}
var (
snapshot *entity.Ticker24h
currentFund *entity.FundingRate
currentOI *entity.OpenInterest
warnMu sync.Mutex
warnings []string
)
addWarn := func(w string) {
warnMu.Lock()
warnings = append(warnings, w)
warnMu.Unlock()
}
g, gctx := errgroup.WithContext(ctx)
g.Go(func() error {
s, err := u.marketData.GetTicker24h(gctx, symbol)
if err != nil {
addWarn("snapshot fetch failed: " + err.Error())
return nil
}
snapshot = s
return nil
})
g.Go(func() error {
f, err := u.derivatives.GetCurrentFunding(gctx, symbol)
if err != nil {
addWarn("current funding fetch failed: " + err.Error())
return nil
}
currentFund = f
return nil
})
g.Go(func() error {
oi, err := u.derivatives.GetCurrentOpenInterest(gctx, symbol)
if err != nil {
addWarn("current OI fetch failed: " + err.Error())
return nil
}
currentOI = oi
return nil
})
_ = g.Wait()
klines := make(map[string][]entity.Kline, len(supportedIntervals))
for _, iv := range supportedIntervals {
rows, err := u.klineRepo.FindRecent(ctx, symbol, iv, klineWindowSize)
if err != nil {
addWarn(fmt.Sprintf("klines DB query failed for %s: %v", iv, err))
rows = nil
}
if len(rows) < klineMinForOK {
addWarn(fmt.Sprintf("klines %s only %d in DB, falling back to Binance", iv, len(rows)))
fresh, ferr := u.marketData.GetKlines(ctx, symbol, iv, klineWindowSize)
if ferr != nil {
addWarn(fmt.Sprintf("klines fallback %s failed: %v", iv, ferr))
} else {
closed := make([]entity.Kline, 0, len(fresh))
for _, k := range fresh {
if k.IsClosed {
closed = append(closed, k)
}
}
go func(items []entity.Kline) {
bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := u.klineRepo.UpsertMany(bgCtx, items); err != nil {
u.log.Error("background_kline_upsert_failed", "err", err)
}
}(closed)
rows = closed
}
}
klines[iv] = rows
}
fundingHist, err := u.fundingRepo.FindRecent(ctx, symbol, fundingHistoryLen)
if err != nil {
addWarn("funding history query failed: " + err.Error())
}
oiHist, err := u.oiRepo.FindRecent(ctx, symbol, derivativePeriod, oiHistoryLen)
if err != nil {
addWarn("OI history query failed: " + err.Error())
}
globalLS, err := u.lsRepo.FindRecent(ctx, symbol, derivativePeriod, entity.RatioTypeGlobalAccount, longShortLen)
if err != nil {
addWarn("global long/short query failed: " + err.Error())
}
topLS, err := u.lsRepo.FindRecent(ctx, symbol, derivativePeriod, entity.RatioTypeTopTraderPosition, longShortLen)
if err != nil {
addWarn("top trader position long/short query failed: " + err.Error())
}
out := &entity.MarketContext{
Symbol: symbol,
GeneratedAt: time.Now().UnixMilli(),
Snapshot: snapshot,
Klines: klines,
Derivatives: entity.DerivativesBundle{
Funding: entity.FundingBundle{
Current: currentFund,
History: fundingHist,
},
OpenInterest: entity.OpenInterestBundle{
Current: currentOI,
History: oiHist,
},
LongShortRatio: entity.LongShortBundle{
Global: globalLS,
TopTraderPosition: topLS,
},
TakerBuySellVolume: nil,
},
Technical: entity.TechnicalStructure{
Support: []entity.TechnicalLevel{},
Resistance: []entity.TechnicalLevel{},
},
DataQuality: entity.DataQuality{
Source: "binance",
Warnings: warnings,
},
}
if out.DataQuality.Warnings == nil {
out.DataQuality.Warnings = []string{}
}
if snapshot == nil {
return out, errors.New("market snapshot unavailable")
}
return out, nil
}

51
internal/usecase/ports.go Normal file
View File

@@ -0,0 +1,51 @@
package usecase
import (
"context"
"cryptoHermes/internal/entity"
)
type MarketDataProvider interface {
GetKlines(ctx context.Context, symbol, interval string, limit int) ([]entity.Kline, error)
GetKlinesRange(ctx context.Context, symbol, interval string, startMs, endMs int64, limit int) ([]entity.Kline, error)
GetTicker24h(ctx context.Context, symbol string) (*entity.Ticker24h, error)
}
type DerivativesProvider interface {
GetCurrentFunding(ctx context.Context, symbol string) (*entity.FundingRate, error)
GetFundingHistory(ctx context.Context, symbol string, limit int) ([]entity.FundingRate, error)
GetCurrentOpenInterest(ctx context.Context, symbol string) (*entity.OpenInterest, error)
GetOpenInterestHistory(ctx context.Context, symbol, period string, limit int) ([]entity.OpenInterest, error)
GetGlobalLongShortRatio(ctx context.Context, symbol, period string, limit int) ([]entity.LongShortRatio, error)
GetTopTraderPositionRatio(ctx context.Context, symbol, period string, limit int) ([]entity.LongShortRatio, error)
GetTakerBuySellVolume(ctx context.Context, symbol, period string, limit int) ([]entity.TakerBuySellVolume, error)
}
type KlineRepository interface {
UpsertMany(ctx context.Context, items []entity.Kline) error
FindRecent(ctx context.Context, symbol, interval string, limit int) ([]entity.Kline, error)
}
type FundingRepository interface {
UpsertMany(ctx context.Context, items []entity.FundingRate) error
FindRecent(ctx context.Context, symbol string, limit int) ([]entity.FundingRate, error)
}
type OpenInterestRepository interface {
UpsertMany(ctx context.Context, items []entity.OpenInterest) error
FindRecent(ctx context.Context, symbol, period string, limit int) ([]entity.OpenInterest, error)
}
type LongShortRatioRepository interface {
UpsertMany(ctx context.Context, items []entity.LongShortRatio) error
FindRecent(ctx context.Context, symbol, period, ratioType string, limit int) ([]entity.LongShortRatio, error)
}
type TakerVolumeRepository interface {
UpsertMany(ctx context.Context, items []entity.TakerBuySellVolume) error
FindRecent(ctx context.Context, symbol, period string, limit int) ([]entity.TakerBuySellVolume, error)
}

152
internal/worker/collector.go Normal file
View File

@@ -0,0 +1,152 @@
package worker
import (
"context"
"log/slog"
"cryptoHermes/internal/entity"
"cryptoHermes/internal/usecase"
)
type Deps struct {
MarketData usecase.MarketDataProvider
Derivatives usecase.DerivativesProvider
KlineRepo usecase.KlineRepository
FundingRepo usecase.FundingRepository
OIRepo usecase.OpenInterestRepository
LSRepo usecase.LongShortRatioRepository
TakerRepo usecase.TakerVolumeRepository
Symbols []string
Intervals []string
Limit int
Logger *slog.Logger
}
type Collector struct {
d Deps
}
func NewCollector(d Deps) *Collector {
if d.Limit <= 0 {
d.Limit = 500
}
return &Collector{d: d}
}
func (c *Collector) Symbols() []string { return c.d.Symbols }
func (c *Collector) Intervals() []string { return c.d.Intervals }
func (c *Collector) CollectKlines(ctx context.Context, interval string) error {
for _, sym := range c.d.Symbols {
ks, err := c.d.MarketData.GetKlines(ctx, sym, interval, c.d.Limit)
if err != nil {
c.d.Logger.Error("collect_klines_failed", "symbol", sym, "interval", interval, "err", err)
continue
}
closed := make([]entity.Kline, 0, len(ks))
for _, k := range ks {
if k.IsClosed {
closed = append(closed, k)
}
}
if err := c.d.KlineRepo.UpsertMany(ctx, closed); err != nil {
c.d.Logger.Error("upsert_klines_failed", "symbol", sym, "interval", interval, "err", err)
continue
}
c.d.Logger.Info("collect_klines_ok", "symbol", sym, "interval", interval, "rows", len(closed))
}
return nil
}
func (c *Collector) CollectAllKlines(ctx context.Context) error {
for _, iv := range c.d.Intervals {
_ = c.CollectKlines(ctx, iv)
}
return nil
}
func (c *Collector) CollectFunding(ctx context.Context) error {
for _, sym := range c.d.Symbols {
hist, err := c.d.Derivatives.GetFundingHistory(ctx, sym, 100)
if err != nil {
c.d.Logger.Error("collect_funding_failed", "symbol", sym, "err", err)
continue
}
if err := c.d.FundingRepo.UpsertMany(ctx, hist); err != nil {
c.d.Logger.Error("upsert_funding_failed", "symbol", sym, "err", err)
continue
}
c.d.Logger.Info("collect_funding_ok", "symbol", sym, "rows", len(hist))
}
return nil
}
func (c *Collector) CollectOpenInterest(ctx context.Context) error {
periods := []string{"5m", "15m", "1h", "4h", "1d"}
for _, sym := range c.d.Symbols {
for _, p := range periods {
hist, err := c.d.Derivatives.GetOpenInterestHistory(ctx, sym, p, 500)
if err != nil {
c.d.Logger.Error("collect_oi_failed", "symbol", sym, "period", p, "err", err)
continue
}
if err := c.d.OIRepo.UpsertMany(ctx, hist); err != nil {
c.d.Logger.Error("upsert_oi_failed", "symbol", sym, "period", p, "err", err)
continue
}
c.d.Logger.Info("collect_oi_ok", "symbol", sym, "period", p, "rows", len(hist))
}
}
return nil
}
func (c *Collector) CollectLongShortRatio(ctx context.Context) error {
periods := []string{"5m", "15m", "1h", "4h", "1d"}
for _, sym := range c.d.Symbols {
for _, p := range periods {
g, err := c.d.Derivatives.GetGlobalLongShortRatio(ctx, sym, p, 500)
if err == nil {
if err := c.d.LSRepo.UpsertMany(ctx, g); err != nil {
c.d.Logger.Error("upsert_global_ls_failed", "symbol", sym, "period", p, "err", err)
} else {
c.d.Logger.Info("collect_global_ls_ok", "symbol", sym, "period", p, "rows", len(g))
}
} else {
c.d.Logger.Error("collect_global_ls_failed", "symbol", sym, "period", p, "err", err)
}
t, err := c.d.Derivatives.GetTopTraderPositionRatio(ctx, sym, p, 500)
if err == nil {
if err := c.d.LSRepo.UpsertMany(ctx, t); err != nil {
c.d.Logger.Error("upsert_top_ls_failed", "symbol", sym, "period", p, "err", err)
} else {
c.d.Logger.Info("collect_top_ls_ok", "symbol", sym, "period", p, "rows", len(t))
}
} else {
c.d.Logger.Error("collect_top_ls_failed", "symbol", sym, "period", p, "err", err)
}
}
}
return nil
}
func (c *Collector) CollectTakerVolume(ctx context.Context) error {
periods := []string{"5m", "15m", "1h", "4h", "1d"}
for _, sym := range c.d.Symbols {
for _, p := range periods {
items, err := c.d.Derivatives.GetTakerBuySellVolume(ctx, sym, p, 500)
if err != nil {
c.d.Logger.Error("collect_taker_failed", "symbol", sym, "period", p, "err", err)
continue
}
if err := c.d.TakerRepo.UpsertMany(ctx, items); err != nil {
c.d.Logger.Error("upsert_taker_failed", "symbol", sym, "period", p, "err", err)
continue
}
c.d.Logger.Info("collect_taker_ok", "symbol", sym, "period", p, "rows", len(items))
}
}
return nil
}