Skip to content

Commit cd3c53f

Browse files
vadiminshakovvadiminshakov
committed
separate events
1 parent bf16c8f commit cd3c53f

2 files changed

Lines changed: 117 additions & 106 deletions

File tree

‎internal/publisher.go‎

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,117 @@
1+
package internal
2+
3+
import (
4+
"context"
5+
"time"
6+
7+
"github.com/pkg/errors"
8+
"github.com/shopspring/decimal"
9+
"go.uber.org/zap"
10+
11+
entity "github.com/vadiminshakov/marti/internal/domain"
12+
)
13+
14+
func (b *TradingBot) streamBalances(ctx context.Context, logger *zap.Logger) {
15+
interval := b.Config.PollPriceInterval
16+
if interval <= 0 {
17+
interval = 5 * time.Second
18+
}
19+
20+
ticker := time.NewTicker(interval)
21+
defer ticker.Stop()
22+
23+
if err := b.publishBalanceSnapshot(ctx); err != nil {
24+
logger.Debug("balance snapshot skipped", zap.Error(err))
25+
}
26+
27+
for {
28+
select {
29+
case <-ctx.Done():
30+
return
31+
case <-ticker.C:
32+
if err := b.publishBalanceSnapshot(ctx); err != nil {
33+
logger.Debug("balance snapshot skipped", zap.Error(err))
34+
}
35+
}
36+
}
37+
}
38+
39+
func (b *TradingBot) publishBalanceSnapshot(ctx context.Context) error {
40+
base, err := b.trader.GetBalance(ctx, b.Config.Pair.From)
41+
if err != nil {
42+
return errors.Wrapf(err, "get %s balance", b.Config.Pair.From)
43+
}
44+
45+
quote, err := b.trader.GetBalance(ctx, b.Config.Pair.To)
46+
if err != nil {
47+
return errors.Wrapf(err, "get %s balance", b.Config.Pair.To)
48+
}
49+
50+
price, err := b.pricer.GetPrice(ctx, b.Config.Pair)
51+
if err != nil {
52+
return errors.Wrap(err, "get price for balance snapshot")
53+
}
54+
55+
total := quote.Add(base.Mul(price))
56+
57+
var (
58+
activePosition string
59+
entryPrice, positionAmount, unrealizedPnL string
60+
)
61+
62+
if b.Config.MarketType == entity.MarketTypeMargin {
63+
position, posErr := b.trader.GetPosition(ctx, b.Config.Pair)
64+
if posErr != nil {
65+
return errors.Wrap(posErr, "get position for balance snapshot")
66+
}
67+
68+
if position != nil && position.Amount.GreaterThan(decimal.Zero) {
69+
switch position.Side {
70+
case entity.PositionSideLong:
71+
activePosition = "long"
72+
case entity.PositionSideShort:
73+
activePosition = "short"
74+
}
75+
76+
total = position.CalculateTotalEquity(price, base, quote, b.leverage)
77+
entryPrice = position.EntryPrice.String()
78+
positionAmount = position.Amount.String()
79+
unrealizedPnL = position.PnL(price).StringFixed(2)
80+
}
81+
}
82+
83+
if b.Config.MarketType == entity.MarketTypeSpot {
84+
if provider, ok := b.tradingStrategy.(DcaCostBasisProvider); ok {
85+
avgPrice, amt := provider.GetDcaCostBasis()
86+
if amt.GreaterThan(decimal.Zero) && avgPrice.GreaterThan(decimal.Zero) {
87+
entryPrice = avgPrice.String()
88+
positionAmount = amt.String()
89+
pnl := price.Sub(avgPrice).Mul(amt)
90+
unrealizedPnL = pnl.StringFixed(2)
91+
activePosition = "long"
92+
}
93+
}
94+
}
95+
96+
model := b.Config.Model
97+
98+
if b.Config.StrategyType == "dca" {
99+
model = "DCA"
100+
}
101+
102+
err = b.balanceStore.Save(entity.NewBalanceSnapshot(
103+
time.Now().UTC(),
104+
b.Config.Pair.String(),
105+
model,
106+
base.String(),
107+
quote.String(),
108+
total.StringFixed(2),
109+
price.String(),
110+
activePosition,
111+
entryPrice,
112+
positionAmount,
113+
unrealizedPnL,
114+
))
115+
116+
return errors.Wrap(err, "failed to save balance snapshot")
117+
}

‎internal/tradingbot.go‎

Lines changed: 0 additions & 106 deletions
Original file line numberDiff line numberDiff line change
@@ -164,110 +164,4 @@ func (b *TradingBot) Run(ctx context.Context, logger *zap.Logger) error {
164164
}
165165
}
166166

167-
func (b *TradingBot) streamBalances(ctx context.Context, logger *zap.Logger) {
168-
interval := b.Config.PollPriceInterval
169-
if interval <= 0 {
170-
interval = 5 * time.Second
171-
}
172-
173-
ticker := time.NewTicker(interval)
174-
defer ticker.Stop()
175-
176-
if err := b.publishBalanceSnapshot(ctx); err != nil {
177-
logger.Debug("balance snapshot skipped", zap.Error(err))
178-
}
179-
180-
for {
181-
select {
182-
case <-ctx.Done():
183-
return
184-
case <-ticker.C:
185-
if err := b.publishBalanceSnapshot(ctx); err != nil {
186-
logger.Debug("balance snapshot skipped", zap.Error(err))
187-
}
188-
}
189-
}
190-
}
191-
192-
func (b *TradingBot) publishBalanceSnapshot(ctx context.Context) error {
193-
base, err := b.trader.GetBalance(ctx, b.Config.Pair.From)
194-
if err != nil {
195-
return errors.Wrapf(err, "get %s balance", b.Config.Pair.From)
196-
}
197-
198-
quote, err := b.trader.GetBalance(ctx, b.Config.Pair.To)
199-
if err != nil {
200-
return errors.Wrapf(err, "get %s balance", b.Config.Pair.To)
201-
}
202-
203-
price, err := b.pricer.GetPrice(ctx, b.Config.Pair)
204-
if err != nil {
205-
return errors.Wrap(err, "get price for balance snapshot")
206-
}
207-
208-
total := quote.Add(base.Mul(price))
209167

210-
var (
211-
activePosition string
212-
entryPrice, positionAmount, unrealizedPnL string
213-
)
214-
215-
if b.Config.MarketType == entity.MarketTypeMargin {
216-
position, posErr := b.trader.GetPosition(ctx, b.Config.Pair)
217-
if posErr != nil {
218-
return errors.Wrap(posErr, "get position for balance snapshot")
219-
}
220-
221-
if position != nil && position.Amount.GreaterThan(decimal.Zero) {
222-
switch position.Side {
223-
case entity.PositionSideLong:
224-
activePosition = "long"
225-
case entity.PositionSideShort:
226-
activePosition = "short"
227-
}
228-
229-
total = position.CalculateTotalEquity(price, base, quote, b.leverage)
230-
entryPrice = position.EntryPrice.String()
231-
positionAmount = position.Amount.String()
232-
unrealizedPnL = position.PnL(price).StringFixed(2)
233-
}
234-
}
235-
236-
// For DCA strategies, get cost basis for PnL calculation.
237-
// Only apply for spot market, as margin positions are handled above via trader.GetPosition().
238-
if b.Config.MarketType == entity.MarketTypeSpot {
239-
if provider, ok := b.tradingStrategy.(DcaCostBasisProvider); ok {
240-
avgPrice, amt := provider.GetDcaCostBasis()
241-
if amt.GreaterThan(decimal.Zero) && avgPrice.GreaterThan(decimal.Zero) {
242-
entryPrice = avgPrice.String()
243-
positionAmount = amt.String()
244-
// PnL for long spot position: (currentPrice - entryPrice) * amount.
245-
pnl := price.Sub(avgPrice).Mul(amt)
246-
unrealizedPnL = pnl.StringFixed(2)
247-
activePosition = "long"
248-
}
249-
}
250-
}
251-
252-
model := b.Config.Model
253-
254-
if b.Config.StrategyType == "dca" {
255-
model = "DCA"
256-
}
257-
258-
err = b.balanceStore.Save(entity.NewBalanceSnapshot(
259-
time.Now().UTC(),
260-
b.Config.Pair.String(),
261-
model,
262-
base.String(),
263-
quote.String(),
264-
total.StringFixed(2),
265-
price.String(),
266-
activePosition,
267-
entryPrice,
268-
positionAmount,
269-
unrealizedPnL,
270-
))
271-
272-
return errors.Wrap(err, "failed to save balance snapshot")
273-
}

0 commit comments

Comments
 (0)