toki
aaf1af6387
Paper trading readiness에서 API/worker/CLI가 같은 protobuf 계약으로 paper state를 시작하고 조회할 수 있어야 한다. Headless 운영 경로를 먼저 닫기 위해 contract, worker runtime, API forwarding, CLI scenario, client parser map과 검증 artifact를 함께 반영한다.
106 lines
3.9 KiB
Go
106 lines
3.9 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"os/signal"
|
|
"syscall"
|
|
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"git.toki-labs.com/toki/alt/packages/domain/market"
|
|
workerbacktest "git.toki-labs.com/toki/alt/services/worker/internal/backtest"
|
|
"git.toki-labs.com/toki/alt/services/worker/internal/config"
|
|
"git.toki-labs.com/toki/alt/services/worker/internal/jobs"
|
|
"git.toki-labs.com/toki/alt/services/worker/internal/marketdata/importer"
|
|
"git.toki-labs.com/toki/alt/services/worker/internal/papertrading"
|
|
"git.toki-labs.com/toki/alt/services/worker/internal/providers/kis"
|
|
"git.toki-labs.com/toki/alt/services/worker/internal/socket"
|
|
"git.toki-labs.com/toki/alt/services/worker/internal/storage/postgres"
|
|
)
|
|
|
|
func main() {
|
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
cfg := config.Load()
|
|
runner := jobs.NewRunner()
|
|
jobs.RegisterBuiltins(runner)
|
|
|
|
// Store-backed socket surfaces are optional at process startup. If the pool
|
|
// cannot be constructed the worker still serves hello, while market/backtest
|
|
// handlers report unavailable instead of panicking.
|
|
var deps socket.Deps
|
|
pool, err := pgxpool.New(ctx, cfg.DatabaseURL)
|
|
if err != nil {
|
|
slog.Error("failed to create database pool; store-backed socket surfaces disabled", "error", err)
|
|
} else {
|
|
defer pool.Close()
|
|
deps = storeBackedDeps(runner, postgres.NewStore(pool))
|
|
}
|
|
|
|
server := socket.NewServer(cfg, deps)
|
|
if err := server.Start(ctx); err != nil {
|
|
slog.Error("failed to start worker socket server", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
slog.Info("worker ready",
|
|
"redis_key_prefix", cfg.RedisKeyPrefix,
|
|
"worker_queue", cfg.WorkerQueue,
|
|
"handlers", runner.Len(),
|
|
"backtest_read_surface", deps.Analysis != nil,
|
|
"backtest_start_surface", deps.Starter != nil,
|
|
"market_read_surface", deps.Instruments != nil && deps.Bars != nil,
|
|
"market_import_surface", deps.DailyBarImporter != nil,
|
|
"addr", fmt.Sprintf("%s:%d%s", cfg.Host, cfg.Port, cfg.SocketPath),
|
|
)
|
|
|
|
<-ctx.Done()
|
|
if err := server.Stop(); err != nil {
|
|
slog.Error("failed to stop worker socket server", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
// storeBackedDeps wires every store-backed socket dependency around a single
|
|
// store, including the backtest execution rail. It is factored out of main so a
|
|
// unit test can assert the backtest start/read surfaces are wired without a live
|
|
// database. Backtest execution reads imported daily bars from storage and runs
|
|
// the bundled strategy resolver; the starter records the pending run and hands
|
|
// execution to the same runner the import jobs use, keeping a single async job
|
|
// rail rather than a second execution engine.
|
|
func storeBackedDeps(runner *jobs.Runner, store *postgres.Store) socket.Deps {
|
|
kisProvider := kis.NewLiveProvider(kis.NewClient(kis.ConfigFromEnv()))
|
|
// One importer instance backs both the async job handler and the socket
|
|
// import command so the operator-facing command and scheduled jobs share the
|
|
// same provider/store wiring.
|
|
kisImporter := importer.New(kisProvider, store)
|
|
jobs.RegisterDailyBarImportHandler(runner, market.ProviderKIS, kisImporter)
|
|
|
|
barSource := workerbacktest.NewStorageBarSource(store, store)
|
|
strategyPort := workerbacktest.NewBuiltInStrategyPort()
|
|
engine := workerbacktest.NewEngine(barSource, strategyPort, store)
|
|
jobs.RegisterRunBacktestHandler(runner, store, engine, time.Now)
|
|
starter := jobs.NewBacktestStarter(runner, store)
|
|
|
|
// Paper trading reuses the same imported-bar source and bundled strategy
|
|
// resolver as backtest, but runs synchronously into a process-local
|
|
// in-memory state keyed by account id (no durable store for readiness).
|
|
paperEngine := papertrading.NewEngine(barSource, strategyPort)
|
|
paperService := papertrading.NewService(paperEngine, time.Now)
|
|
|
|
return socket.Deps{
|
|
Starter: starter,
|
|
Analysis: store,
|
|
Results: store,
|
|
Instruments: store,
|
|
Bars: store,
|
|
DailyBarImporter: kisImporter,
|
|
Paper: paperService,
|
|
}
|
|
}
|