lychiyu/bbgo_origin
1
0
Fork 0
mirror of https://github.com/c9s/bbgo.git synced 2024-11-10 09:11:55 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #997 from zenixls2/fix/serialMarketDataStore

Browse Source
This commit is contained in:
Yo-An Lin 2022-10-31 18:00:39 +08:00 committed by GitHub
commit 999d7b3799
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 757 additions and 617 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

102
config/drift.yaml
View File

@ -1,102 +0,0 @@
---
persistence:
redis:
host: 127.0.0.1
port: 6379
db: 0
sessions:
binance:
exchange: binance
futures: false
envVarPrefix: binance
heikinAshi: false
# Drift strategy intends to place buy/sell orders as much value mas it could be. To exchanges that requires to
# calculate fees before placing limit orders (e.g. FTX Pro), make sure the fee rate is configured correctly and
# enable modifyOrderAmountForFee to prevent order rejection.
makerFeeRate: 0.0002
takerFeeRate: 0.0007
modifyOrderAmountForFee: false
exchangeStrategies:
- on: binance
drift:
canvasPath: "./output.png"
symbol: ETHBUSD
limitOrder: false
quantity: 0.01
# kline interval for indicators
interval: 1m
window: 1
useAtr: true
useStopLoss: true
stoploss: 0.23%
source: ohlc4
predictOffset: 2
noTrailingStopLoss: false
trailingStopLossType: kline
# stddev on high/low-source
hlVarianceMultiplier: 0.13
hlRangeWindow: 4
smootherWindow: 19
fisherTransformWindow: 73
atrWindow: 14
# orders not been traded will be canceled after `pendingMinutes` minutes
pendingMinutes: 5
noRebalance: true
trendWindow: 12
rebalanceFilter: 2
trailingActivationRatio: [0.0015, 0.002, 0.004, 0.01]
trailingCallbackRate: [0.0001, 0.00012, 0.001, 0.002]
generateGraph: true
graphPNLDeductFee: true
graphPNLPath: "./pnl.png"
graphCumPNLPath: "./cumpnl.png"
#exits:
#- roiStopLoss:
# percentage: 0.8%
#- roiTakeProfit:
# percentage: 35%
#- protectiveStopLoss:
# activationRatio: 0.6%
# stopLossRatio: 0.1%
# placeStopOrder: false
#- protectiveStopLoss:
# activationRatio: 5%
# stopLossRatio: 1%
# placeStopOrder: false
#- cumulatedVolumeTakeProfit:
# interval: 5m
# window: 2
# minQuoteVolume: 200_000_000
#- protectiveStopLoss:
# activationRatio: 2%
# stopLossRatio: 1%
# placeStopOrder: false
sync:
userDataStream:
trades: true
filledOrders: true
sessions:
- binance
symbols:
- ETHBUSD
backtest:
startTime: "2022-09-25"
endTime: "2022-09-30"
symbols:
- ETHBUSD
sessions: [binance]
accounts:
binance:
makerFeeRate: 0.0000
takerFeeRate: 0.0000
balances:
ETH: 0.03
BUSD: 0

32
config/driftBTC.yaml
View File

@ -26,32 +26,32 @@ exchangeStrategies:
- on: binance
drift:
debug: false
minInterval: 1s
limitOrder: true
#quantity: 0.0012
canvasPath: "./output.png"
symbol: BTCUSDT
# kline interval for indicators
interval: 1m
window: 6
interval: 1s
window: 2
useAtr: true
useStopLoss: true
stoploss: 0.05%
stoploss: 0.01%
source: hl2
predictOffset: 2
noTrailingStopLoss: false
trailingStopLossType: kline
noTrailingStopLoss: true
# stddev on high/low-source
hlVarianceMultiplier: 0.14
hlRangeWindow: 4
smootherWindow: 3
fisherTransformWindow: 125
#fisherTransformWindow: 117
hlVarianceMultiplier: 0.7
hlRangeWindow: 6
smootherWindow: 10
fisherTransformWindow: 45
atrWindow: 24
# orders not been traded will be canceled after `pendingMinutes` minutes
pendingMinutes: 10
pendingMinInterval: 6
noRebalance: true
trendWindow: 15
rebalanceFilter: -0.1
trendWindow: 4
rebalanceFilter: 2
# ActivationRatio should be increasing order
# when farest price from entry goes over that ratio, start using the callback ratio accordingly to do trailingstop
@ -68,6 +68,7 @@ exchangeStrategies:
graphPNLDeductFee: false
graphPNLPath: "./pnl.png"
graphCumPNLPath: "./cumpnl.png"
graphElapsedPath: "./elapsed.png"
#exits:
# - roiStopLoss:
# percentage: 0.35%
@ -125,11 +126,12 @@ sync:
- BTCUSDT
backtest:
startTime: "2022-09-25"
endTime: "2022-10-30"
startTime: "2022-10-19"
endTime: "2022-10-20"
symbols:
- BTCUSDT
sessions: [binance]
syncSecKLines: true
accounts:
binance:
makerFeeRate: 0.000

34
config/elliottwave.yaml
View File

@ -23,17 +23,18 @@ exchangeStrategies:
- on: binance
elliottwave:
symbol: BNBBUSD
minInterval: 1s
symbol: BTCUSDT
limitOrder: true
quantity: 0.16
#quantity: 0.16
# kline interval for indicators
interval: 1m
interval: 1s
stoploss: 0.01%
windowATR: 14
windowQuick: 5
windowSlow: 9
windowQuick: 4
windowSlow: 155
source: hl2
pendingMinutes: 10
pendingMinInterval: 5
useHeikinAshi: true
drawGraph: true
@ -46,12 +47,12 @@ exchangeStrategies:
# when farest price from entry goes over that ratio, start using the callback ratio accordingly to do trailingstop
#trailingActivationRatio: [0.01, 0.016, 0.05]
#trailingActivationRatio: [0.001, 0.0081, 0.022]
trailingActivationRatio: [0.0017, 0.01, 0.015]
#trailingActivationRatio: []
#trailingCallbackRate: []
#trailingActivationRatio: [0.0017, 0.01, 0.015]
trailingActivationRatio: []
trailingCallbackRate: []
#trailingCallbackRate: [0.002, 0.01, 0.1]
#trailingCallbackRate: [0.0004, 0.0009, 0.018]
trailingCallbackRate: [0.0006, 0.0019, 0.006]
#trailingCallbackRate: [0.0006, 0.0019, 0.006]
#exits:
# - roiStopLoss:
@ -107,18 +108,19 @@ sync:
sessions:
- binance
symbols:
- BNBBUSD
- BTCUSDT
backtest:
startTime: "2022-09-01"
endTime: "2022-09-30"
startTime: "2022-10-15"
endTime: "2022-10-19"
symbols:
- BNBBUSD
- BTCUSDT
sessions: [binance]
syncSecKLines: true
accounts:
binance:
makerFeeRate: 0.000
takerFeeRate: 0.000
balances:
BNB: 0
BUSD: 100
BTC: 0
USDT: 100

2
pkg/bbgo/marketdatastore.go
View File

@ -22,7 +22,7 @@ func NewMarketDataStore(symbol string) *MarketDataStore {
Symbol: symbol,
// KLineWindows stores all loaded klines per interval
KLineWindows: make(map[types.Interval]*types.KLineWindow, len(types.SupportedIntervals)), // 12 interval, 1m,5m,15m,30m,1h,2h,4h,6h,12h,1d,3d,1w
KLineWindows: make(map[types.Interval]*types.KLineWindow, len(types.SupportedIntervals)), // 13 interval, 1s,1m,5m,15m,30m,1h,2h,4h,6h,12h,1d,3d,1w
}
}

127
pkg/bbgo/serialmarketdatastore.go
View File

@ -1,22 +1,35 @@
package bbgo
import (
"context"
"sync"
"time"
"github.com/c9s/bbgo/pkg/fixedpoint"
"github.com/c9s/bbgo/pkg/types"
log "github.com/sirupsen/logrus"
)
type SerialMarketDataStore struct {
*MarketDataStore
KLines map[types.Interval]*types.KLine
Subscription []types.Interval
UseMarketTrade bool
KLines map[types.Interval]*types.KLine
MinInterval types.Interval
Subscription []types.Interval
o, h, l, c, v, qv, price fixedpoint.Value
mu sync.Mutex
}
func NewSerialMarketDataStore(symbol string) *SerialMarketDataStore {
// @param symbol: symbol to trace on
// @param minInterval: unit interval, related to your signal timeframe
// @param useMarketTrade: if not assigned, default to false. if assigned to true, will use MarketTrade signal to generate klines
func NewSerialMarketDataStore(symbol string, minInterval types.Interval, useMarketTrade ...bool) *SerialMarketDataStore {
return &SerialMarketDataStore{
MarketDataStore: NewMarketDataStore(symbol),
KLines: make(map[types.Interval]*types.KLine),
UseMarketTrade: len(useMarketTrade) > 0 && useMarketTrade[0],
Subscription: []types.Interval{},
MinInterval: minInterval,
}
}
@ -30,24 +43,110 @@ func (store *SerialMarketDataStore) Subscribe(interval types.Interval) {
store.Subscription = append(store.Subscription, interval)
}
func (store *SerialMarketDataStore) BindStream(stream types.Stream) {
stream.OnKLineClosed(store.handleKLineClosed)
func (store *SerialMarketDataStore) BindStream(ctx context.Context, stream types.Stream) {
if store.UseMarketTrade {
if IsBackTesting {
log.Errorf("right now in backtesting, aggTrade event is not yet supported. Use OnKLineClosed instead.")
stream.OnKLineClosed(store.handleKLineClosed)
return
}
go store.tickerProcessor(ctx)
stream.OnMarketTrade(store.handleMarketTrade)
} else {
stream.OnKLineClosed(store.handleKLineClosed)
}
}
func (store *SerialMarketDataStore) handleKLineClosed(kline types.KLine) {
store.AddKLine(kline)
}
func (store *SerialMarketDataStore) AddKLine(kline types.KLine) {
func (store *SerialMarketDataStore) handleMarketTrade(trade types.Trade) {
store.mu.Lock()
store.price = trade.Price
store.c = store.price
if store.price.Compare(store.h) > 0 {
store.h = store.price
}
if !store.l.IsZero() {
if store.price.Compare(store.l) < 0 {
store.l = store.price
}
} else {
store.l = store.price
}
if store.o.IsZero() {
store.o = store.price
}
store.v = store.v.Add(trade.Quantity)
store.qv = store.qv.Add(trade.QuoteQuantity)
store.mu.Unlock()
}
func (store *SerialMarketDataStore) tickerProcessor(ctx context.Context) {
duration := store.MinInterval.Duration()
relativeTime := time.Now().UnixNano() % int64(duration)
waitTime := int64(duration) - relativeTime
select {
case <-time.After(time.Duration(waitTime)):
case <-ctx.Done():
return
}
intervalCloseTicker := time.NewTicker(duration)
defer intervalCloseTicker.Stop()
for {
select {
case time := <-intervalCloseTicker.C:
kline := types.KLine{
Symbol: store.Symbol,
StartTime: types.Time(time.Add(-1 * duration).Round(duration)),
EndTime: types.Time(time),
Interval: store.MinInterval,
Closed: true,
}
store.mu.Lock()
if store.c.IsZero() {
kline.Open = store.price
kline.Close = store.price
kline.High = store.price
kline.Low = store.price
kline.Volume = fixedpoint.Zero
kline.QuoteVolume = fixedpoint.Zero
} else {
kline.Open = store.o
kline.Close = store.c
kline.High = store.h
kline.Low = store.l
kline.Volume = store.v
kline.QuoteVolume = store.qv
store.o = fixedpoint.Zero
store.c = fixedpoint.Zero
store.h = fixedpoint.Zero
store.l = fixedpoint.Zero
store.v = fixedpoint.Zero
store.qv = fixedpoint.Zero
}
store.mu.Unlock()
store.AddKLine(kline, true)
case <-ctx.Done():
return
}
}
}
func (store *SerialMarketDataStore) AddKLine(kline types.KLine, async ...bool) {
if kline.Symbol != store.Symbol {
return
}
// only consumes kline1m
if kline.Interval != types.Interval1m {
// only consumes MinInterval
if kline.Interval != store.MinInterval {
return
}
// endtime in minutes
timestamp := kline.StartTime.Time().Add(time.Minute)
// endtime
duration := store.MinInterval.Duration()
timestamp := kline.StartTime.Time().Add(duration)
for _, val := range store.Subscription {
k, ok := store.KLines[val]
if !ok {
@ -60,9 +159,13 @@ func (store *SerialMarketDataStore) AddKLine(kline types.KLine) {
k.Merge(&kline)
k.Closed = false
}
if timestamp.Truncate(val.Duration()) == timestamp {
if timestamp.Round(val.Duration()) == timestamp {
k.Closed = true
store.MarketDataStore.AddKLine(*k)
if len(async) > 0 && async[0] {
go store.MarketDataStore.AddKLine(*k)
} else {
store.MarketDataStore.AddKLine(*k)
}
delete(store.KLines, val)
}
}

29
pkg/bbgo/session.go
View File

@ -395,8 +395,7 @@ func (session *ExchangeSession) initSymbol(ctx context.Context, environ *Environ
// used kline intervals by the given symbol
var klineSubscriptions = map[types.Interval]struct{}{}
// always subscribe the 1m kline so we can make sure the connection persists.
klineSubscriptions[types.Interval1m] = struct{}{}
minInterval := types.Interval1m
// Aggregate the intervals that we are using in the subscriptions.
for _, sub := range session.Subscriptions {
@ -411,12 +410,19 @@ func (session *ExchangeSession) initSymbol(ctx context.Context, environ *Environ
continue
}
if minInterval.Seconds() > sub.Options.Interval.Seconds() {
minInterval = sub.Options.Interval
}
if sub.Symbol == symbol {
klineSubscriptions[types.Interval(sub.Options.Interval)] = struct{}{}
}
}
}
// always subscribe the 1m kline so we can make sure the connection persists.
klineSubscriptions[minInterval] = struct{}{}
for interval := range klineSubscriptions {
// avoid querying the last unclosed kline
endTime := environ.startTime
@ -440,7 +446,7 @@ func (session *ExchangeSession) initSymbol(ctx context.Context, environ *Environ
// update last prices by the given kline
lastKLine := kLines[len(kLines)-1]
if interval == types.Interval1m {
if interval == minInterval {
session.lastPrices[symbol] = lastKLine.Close
}
@ -497,6 +503,7 @@ func (session *ExchangeSession) Positions() map[string]*types.Position {
// MarketDataStore returns the market data store of a symbol
func (session *ExchangeSession) MarketDataStore(symbol string) (s *MarketDataStore, ok bool) {
s, ok = session.marketDataStores[symbol]
// FIXME: the returned MarketDataStore when !ok will be empty
if !ok {
s = NewMarketDataStore(symbol)
s.BindStream(session.MarketDataStream)
@ -507,15 +514,21 @@ func (session *ExchangeSession) MarketDataStore(symbol string) (s *MarketDataSto
}
// KLine updates will be received in the order listend in intervals array
func (session *ExchangeSession) SerialMarketDataStore(symbol string, intervals []types.Interval) (store *SerialMarketDataStore, ok bool) {
func (session *ExchangeSession) SerialMarketDataStore(ctx context.Context, symbol string, intervals []types.Interval, useAggTrade ...bool) (store *SerialMarketDataStore, ok bool) {
st, ok := session.MarketDataStore(symbol)
if !ok {
return nil, false
}
store = NewSerialMarketDataStore(symbol)
klines, ok := st.KLinesOfInterval(types.Interval1m)
minInterval := types.Interval1m
for _, i := range intervals {
if minInterval.Seconds() > i.Seconds() {
minInterval = i
}
}
store = NewSerialMarketDataStore(symbol, minInterval, useAggTrade...)
klines, ok := st.KLinesOfInterval(minInterval)
if !ok {
log.Errorf("SerialMarketDataStore: cannot get 1m history")
log.Errorf("SerialMarketDataStore: cannot get %s history", minInterval)
return nil, false
}
for _, interval := range intervals {
@ -524,7 +537,7 @@ func (session *ExchangeSession) SerialMarketDataStore(symbol string, intervals [
for _, kline := range *klines {
store.AddKLine(kline)
}
store.BindStream(session.MarketDataStream)
store.BindStream(ctx, session.MarketDataStream)
return store, true
}

94
pkg/bbgo/trader.go
View File

@ -166,52 +166,6 @@ func (trader *Trader) SetRiskControls(riskControls *RiskControls) {
trader.riskControls = riskControls
}
func (trader *Trader) Subscribe() {
// pre-subscribe the data
for sessionName, strategies := range trader.exchangeStrategies {
session := trader.environment.sessions[sessionName]
for _, strategy := range strategies {
if defaulter, ok := strategy.(StrategyDefaulter); ok {
if err := defaulter.Defaults(); err != nil {
panic(err)
}
}
if initializer, ok := strategy.(StrategyInitializer); ok {
if err := initializer.Initialize(); err != nil {
panic(err)
}
}
if subscriber, ok := strategy.(ExchangeSessionSubscriber); ok {
subscriber.Subscribe(session)
} else {
log.Errorf("strategy %s does not implement ExchangeSessionSubscriber", strategy.ID())
}
}
}
for _, strategy := range trader.crossExchangeStrategies {
if defaulter, ok := strategy.(StrategyDefaulter); ok {
if err := defaulter.Defaults(); err != nil {
panic(err)
}
}
if initializer, ok := strategy.(StrategyInitializer); ok {
if err := initializer.Initialize(); err != nil {
panic(err)
}
}
if subscriber, ok := strategy.(CrossExchangeSessionSubscriber); ok {
subscriber.CrossSubscribe(trader.environment.sessions)
} else {
log.Errorf("strategy %s does not implement CrossExchangeSessionSubscriber", strategy.ID())
}
}
}
func (trader *Trader) RunSingleExchangeStrategy(ctx context.Context, strategy SingleExchangeStrategy, session *ExchangeSession, orderExecutor OrderExecutor) error {
if v, ok := strategy.(StrategyValidator); ok {
if err := v.Validate(); err != nil {
@ -262,7 +216,7 @@ func (trader *Trader) RunAllSingleExchangeStrategy(ctx context.Context) error {
return nil
}
func (trader *Trader) injectFields() error {
func (trader *Trader) injectFieldsAndSubscribe(ctx context.Context) error {
// load and run Session strategies
for sessionName, strategies := range trader.exchangeStrategies {
var session = trader.environment.sessions[sessionName]
@ -285,8 +239,30 @@ func (trader *Trader) injectFields() error {
return errors.Wrapf(err, "failed to inject OrderExecutor on %T", strategy)
}
if defaulter, ok := strategy.(StrategyDefaulter); ok {
if err := defaulter.Defaults(); err != nil {
panic(err)
}
}
if initializer, ok := strategy.(StrategyInitializer); ok {
if err := initializer.Initialize(); err != nil {
panic(err)
}
}
if subscriber, ok := strategy.(ExchangeSessionSubscriber); ok {
subscriber.Subscribe(session)
} else {
log.Errorf("strategy %s does not implement ExchangeSessionSubscriber", strategy.ID())
}
if symbol, ok := dynamic.LookupSymbolField(rs); ok {
log.Infof("found symbol based strategy from %s", rs.Type())
log.Infof("found symbol %s based strategy from %s", symbol, rs.Type())
if err := session.initSymbol(ctx, trader.environment, symbol); err != nil {
return errors.Wrapf(err, "failed to inject object into %T when initSymbol", strategy)
}
market, ok := session.Market(symbol)
if !ok {
@ -328,6 +304,24 @@ func (trader *Trader) injectFields() error {
if err := trader.injectCommonServices(strategy); err != nil {
return err
}
if defaulter, ok := strategy.(StrategyDefaulter); ok {
if err := defaulter.Defaults(); err != nil {
return err
}
}
if initializer, ok := strategy.(StrategyInitializer); ok {
if err := initializer.Initialize(); err != nil {
return err
}
}
if subscriber, ok := strategy.(CrossExchangeSessionSubscriber); ok {
subscriber.CrossSubscribe(trader.environment.sessions)
} else {
log.Errorf("strategy %s does not implement CrossExchangeSessionSubscriber", strategy.ID())
}
}
return nil
@ -339,12 +333,10 @@ func (trader *Trader) Run(ctx context.Context) error {
// trader.environment.Connect will call interact.Start
interact.AddCustomInteraction(NewCoreInteraction(trader.environment, trader))
if err := trader.injectFields(); err != nil {
if err := trader.injectFieldsAndSubscribe(ctx); err != nil {
return err
}
trader.Subscribe()
if err := trader.environment.Start(ctx); err != nil {
return err
}

4
pkg/exchange/binance/parse.go
View File

@ -538,7 +538,7 @@ func (e *MarketTradeEvent) Trade() types.Trade {
Side: side,
Price: e.Price,
Quantity: e.Quantity,
QuoteQuantity: e.Quantity,
QuoteQuantity: e.Quantity.Mul(e.Price),
IsBuyer: isBuyer,
IsMaker: e.IsMaker,
Time: types.Time(tt),
@ -595,7 +595,7 @@ func (e *AggTradeEvent) Trade() types.Trade {
Side: side,
Price: e.Price,
Quantity: e.Quantity,
QuoteQuantity: e.Quantity,
QuoteQuantity: e.Quantity.Mul(e.Price),
IsBuyer: isBuyer,
IsMaker: e.IsMaker,
Time: types.Time(tt),

6
pkg/indicator/atr.go
View File

@ -8,6 +8,9 @@ import (
"github.com/c9s/bbgo/pkg/types"
)
const MaxNumOfATR = 1000
const MaxNumOfATRTruncateSize = 500
//go:generate callbackgen -type ATR
type ATR struct {
types.SeriesBase
@ -73,6 +76,9 @@ func (inc *ATR) Update(high, low, cloze float64) {
inc.RMA.Update(trueRange)
atr := inc.RMA.Last()
inc.PercentageVolatility.Push(atr / cloze)
if len(inc.PercentageVolatility) > MaxNumOfATR {
inc.PercentageVolatility = inc.PercentageVolatility[MaxNumOfATRTruncateSize-1:]
}
}
func (inc *ATR) Last() float64 {

4
pkg/indicator/ewma.go
View File

@ -8,8 +8,8 @@ import (
)
// These numbers should be aligned with bbgo MaxNumOfKLines and MaxNumOfKLinesTruncate
const MaxNumOfEWMA = 5_000
const MaxNumOfEWMATruncateSize = 100
const MaxNumOfEWMA = 1_000
const MaxNumOfEWMATruncateSize = 500
//go:generate callbackgen -type EWMA
type EWMA struct {

6
pkg/indicator/rma.go
View File

@ -7,6 +7,9 @@ import (
"github.com/c9s/bbgo/pkg/types"
)
const MaxNumOfRMA = 1000
const MaxNumOfRMATruncateSize = 500
// Running Moving Average
// Refer: https://github.com/twopirllc/pandas-ta/blob/main/pandas_ta/overlap/rma.py#L5
// Refer: https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.ewm.html#pandas-dataframe-ewm
@ -62,6 +65,9 @@ func (inc *RMA) Update(x float64) {
}
inc.Values.Push(inc.tmp)
if len(inc.Values) > MaxNumOfRMA {
inc.Values = inc.Values[MaxNumOfRMATruncateSize-1:]
}
}
func (inc *RMA) Last() float64 {

3
pkg/indicator/sma.go
View File

@ -65,6 +65,9 @@ func (inc *SMA) Update(value float64) {
}
inc.Values.Push(types.Mean(inc.rawValues))
if len(inc.Values) > MaxNumOfSMA {
inc.Values = inc.Values[MaxNumOfSMATruncateSize-1:]
}
}
func (inc *SMA) BindK(target KLineClosedEmitter, symbol string, interval types.Interval) {

6
pkg/indicator/stddev.go
View File

@ -7,6 +7,9 @@ import (
"github.com/c9s/bbgo/pkg/types"
)
const MaxNumOfStdev = 600
const MaxNumOfStdevTruncateSize = 300
//go:generate callbackgen -type StdDev
type StdDev struct {
types.SeriesBase
@ -49,6 +52,9 @@ func (inc *StdDev) Update(value float64) {
var std = inc.rawValues.Stdev()
inc.Values.Push(std)
if len(inc.Values) > MaxNumOfStdev {
inc.Values = inc.Values[MaxNumOfStdevTruncateSize-1:]
}
}
func (inc *StdDev) PushK(k types.KLine) {

2
pkg/notifier/telegramnotifier/telegram.go
View File

@ -15,7 +15,7 @@ import (
"github.com/c9s/bbgo/pkg/types"
)
var apiLimiter = rate.NewLimiter(rate.Every(1*time.Second), 1)
var apiLimiter = rate.NewLimiter(rate.Every(50*time.Millisecond), 20)
var log = logrus.WithField("service", "telegram")

175
pkg/strategy/drift/draw.go Normal file
View File

@ -0,0 +1,175 @@
package drift
import (
"bytes"
"fmt"
"os"
"github.com/c9s/bbgo/pkg/bbgo"
"github.com/c9s/bbgo/pkg/interact"
"github.com/c9s/bbgo/pkg/types"
"github.com/wcharczuk/go-chart/v2"
)
func (s *Strategy) InitDrawCommands(profit, cumProfit types.Series) {
bbgo.RegisterCommand("/draw", "Draw Indicators", func(reply interact.Reply) {
go func() {
canvas := s.DrawIndicators(s.frameKLine.StartTime)
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render indicators in drift")
return
}
bbgo.SendPhoto(&buffer)
}()
})
bbgo.RegisterCommand("/pnl", "Draw PNL(%) per trade", func(reply interact.Reply) {
go func() {
canvas := s.DrawPNL(profit)
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render pnl in drift")
return
}
bbgo.SendPhoto(&buffer)
}()
})
bbgo.RegisterCommand("/cumpnl", "Draw Cummulative PNL(Quote)", func(reply interact.Reply) {
go func() {
canvas := s.DrawCumPNL(cumProfit)
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render cumpnl in drift")
return
}
bbgo.SendPhoto(&buffer)
}()
})
bbgo.RegisterCommand("/elapsed", "Draw Elapsed time for handlers for each kline close event", func(reply interact.Reply) {
go func() {
canvas := s.DrawElapsed()
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render elapsed in drift")
return
}
bbgo.SendPhoto(&buffer)
}()
})
}
func (s *Strategy) DrawIndicators(time types.Time) *types.Canvas {
canvas := types.NewCanvas(s.InstanceID(), s.Interval)
length := s.priceLines.Length()
if length > 300 {
length = 300
}
log.Infof("draw indicators with %d data", length)
mean := s.priceLines.Mean(length)
highestPrice := s.priceLines.Minus(mean).Abs().Highest(length)
highestDrift := s.drift.Abs().Highest(length)
hi := s.drift.drift.Abs().Highest(length)
ratio := highestPrice / highestDrift
// canvas.Plot("upband", s.ma.Add(s.stdevHigh), time, length)
canvas.Plot("ma", s.ma, time, length)
// canvas.Plot("downband", s.ma.Minus(s.stdevLow), time, length)
fmt.Printf("%f %f\n", highestPrice, hi)
canvas.Plot("trend", s.trendLine, time, length)
canvas.Plot("drift", s.drift.Mul(ratio).Add(mean), time, length)
canvas.Plot("driftOrig", s.drift.drift.Mul(highestPrice/hi).Add(mean), time, length)
canvas.Plot("zero", types.NumberSeries(mean), time, length)
canvas.Plot("price", s.priceLines, time, length)
return canvas
}
func (s *Strategy) DrawPNL(profit types.Series) *types.Canvas {
canvas := types.NewCanvas(s.InstanceID())
log.Errorf("pnl Highest: %f, Lowest: %f", types.Highest(profit, profit.Length()), types.Lowest(profit, profit.Length()))
length := profit.Length()
if s.GraphPNLDeductFee {
canvas.PlotRaw("pnl % (with Fee Deducted)", profit, length)
} else {
canvas.PlotRaw("pnl %", profit, length)
}
canvas.YAxis = chart.YAxis{
ValueFormatter: func(v interface{}) string {
if vf, isFloat := v.(float64); isFloat {
return fmt.Sprintf("%.4f", vf)
}
return ""
},
}
canvas.PlotRaw("1", types.NumberSeries(1), length)
return canvas
}
func (s *Strategy) DrawCumPNL(cumProfit types.Series) *types.Canvas {
canvas := types.NewCanvas(s.InstanceID())
canvas.PlotRaw("cummulative pnl", cumProfit, cumProfit.Length())
canvas.YAxis = chart.YAxis{
ValueFormatter: func(v interface{}) string {
if vf, isFloat := v.(float64); isFloat {
return fmt.Sprintf("%.4f", vf)
}
return ""
},
}
return canvas
}
func (s *Strategy) DrawElapsed() *types.Canvas {
canvas := types.NewCanvas(s.InstanceID())
canvas.PlotRaw("elapsed time(ms)", s.elapsed, s.elapsed.Length())
return canvas
}
func (s *Strategy) Draw(time types.Time, profit types.Series, cumProfit types.Series) {
canvas := s.DrawIndicators(time)
f, err := os.Create(s.CanvasPath)
if err != nil {
log.WithError(err).Errorf("cannot create on %s", s.CanvasPath)
return
}
if err := canvas.Render(chart.PNG, f); err != nil {
log.WithError(err).Errorf("cannot render in drift")
}
f.Close()
canvas = s.DrawPNL(profit)
f, err = os.Create(s.GraphPNLPath)
if err != nil {
log.WithError(err).Errorf("open pnl")
return
}
if err := canvas.Render(chart.PNG, f); err != nil {
log.WithError(err).Errorf("render pnl")
}
f.Close()
canvas = s.DrawCumPNL(cumProfit)
f, err = os.Create(s.GraphCumPNLPath)
if err != nil {
log.WithError(err).Errorf("open cumpnl")
return
}
if err := canvas.Render(chart.PNG, f); err != nil {
log.WithError(err).Errorf("render cumpnl")
}
f.Close()
canvas = s.DrawElapsed()
f, err = os.Create(s.GraphElapsedPath)
if err != nil {
log.WithError(err).Errorf("open elapsed")
return
}
if err := canvas.Render(chart.PNG, f); err != nil {
log.WithError(err).Errorf("render elapsed")
}
f.Close()
}

510
pkg/strategy/drift/strategy.go
View File

@ -12,7 +12,6 @@ import (
"time"
"github.com/sirupsen/logrus"
"github.com/wcharczuk/go-chart/v2"
"go.uber.org/multierr"
"github.com/c9s/bbgo/pkg/bbgo"
@ -65,57 +64,61 @@ type Strategy struct {
*types.ProfitStats `persistence:"profit_stats"`
*types.TradeStats `persistence:"trade_stats"`
p *types.Position
p *types.Position
MinInterval types.Interval `json:"MinInterval"` // minimum interval referred for doing stoploss/trailing exists and updating highest/lowest
priceLines *types.Queue
trendLine types.UpdatableSeriesExtend
ma types.UpdatableSeriesExtend
stdevHigh *indicator.StdDev
stdevLow *indicator.StdDev
drift *DriftMA
atr *indicator.ATR
midPrice fixedpoint.Value
lock sync.RWMutex `ignore:"true"`
positionLock sync.RWMutex `ignore:"true"`
startTime time.Time
minutesCounter int
orderPendingCounter map[uint64]int
frameKLine *types.KLine
kline1m *types.KLine
elapsed *types.Queue
priceLines *types.Queue
trendLine types.UpdatableSeriesExtend
ma types.UpdatableSeriesExtend
stdevHigh *indicator.StdDev
stdevLow *indicator.StdDev
drift *DriftMA
atr *indicator.ATR
midPrice fixedpoint.Value // the midPrice is the average of bestBid and bestAsk in public orderbook
lock sync.RWMutex `ignore:"true"` // lock for midPrice
positionLock sync.RWMutex `ignore:"true"` // lock for highest/lowest and p
pendingLock sync.Mutex `ignore:"true"`
startTime time.Time // trading start time
maxCounterBuyCanceled int // the largest counter of the order on the buy side been cancelled. meaning the latest cancelled buy order.
maxCounterSellCanceled int // the largest counter of the order on the sell side been cancelled. meaning the latest cancelled sell order.
orderPendingCounter map[uint64]int // records the timepoint when the orders are created, using the counter at the time.
frameKLine *types.KLine // last kline in Interval
klineMin *types.KLine // last kline in MinInterval
beta float64
beta float64 // last beta value from trendline's linear regression (previous slope of the trendline)
UseStopLoss bool `json:"useStopLoss" modifiable:"true"`
UseAtr bool `json:"useAtr" modifiable:"true"`
StopLoss fixedpoint.Value `json:"stoploss" modifiable:"true"`
CanvasPath string `json:"canvasPath"`
PredictOffset int `json:"predictOffset"`
HighLowVarianceMultiplier float64 `json:"hlVarianceMultiplier" modifiable:"true"`
NoTrailingStopLoss bool `json:"noTrailingStopLoss" modifiable:"true"`
TrailingStopLossType string `json:"trailingStopLossType" modifiable:"true"` // trailing stop sources. Possible options are `kline` for 1m kline and `realtime` from order updates
HLRangeWindow int `json:"hlRangeWindow"`
SmootherWindow int `json:"smootherWindow"`
FisherTransformWindow int `json:"fisherTransformWindow"`
ATRWindow int `json:"atrWindow"`
PendingMinutes int `json:"pendingMinutes" modifiable:"true"` // if order not be traded for pendingMinutes of time, cancel it.
NoRebalance bool `json:"noRebalance" modifiable:"true"` // disable rebalance
TrendWindow int `json:"trendWindow"` // trendLine is used for rebalancing the position. When trendLine goes up, hold base, otherwise hold quote
RebalanceFilter float64 `json:"rebalanceFilter" modifiable:"true"` // beta filter on the Linear Regression of trendLine
Debug bool `json:"debug" modifiable:"true"` // to print debug message or not
UseStopLoss bool `json:"useStopLoss" modifiable:"true"` // whether to use stoploss rate to do stoploss
UseAtr bool `json:"useAtr" modifiable:"true"` // use atr as stoploss
StopLoss fixedpoint.Value `json:"stoploss" modifiable:"true"` // stoploss rate
PredictOffset int `json:"predictOffset"` // the lookback length for the prediction using linear regression
HighLowVarianceMultiplier float64 `json:"hlVarianceMultiplier" modifiable:"true"` // modifier to set the limit order price
NoTrailingStopLoss bool `json:"noTrailingStopLoss" modifiable:"true"` // turn off the trailing exit and stoploss
HLRangeWindow int `json:"hlRangeWindow"` // ma window for kline high/low changes
SmootherWindow int `json:"smootherWindow"` // window that controls the smoothness of drift
FisherTransformWindow int `json:"fisherTransformWindow"` // fisher transform window to filter drift's negative signals
ATRWindow int `json:"atrWindow"` // window for atr indicator
PendingMinInterval int `json:"pendingMinInterval" modifiable:"true"` // if order not be traded for pendingMinInterval of time, cancel it.
NoRebalance bool `json:"noRebalance" modifiable:"true"` // disable rebalance
TrendWindow int `json:"trendWindow"` // trendLine is used for rebalancing the position. When trendLine goes up, hold base, otherwise hold quote
RebalanceFilter float64 `json:"rebalanceFilter" modifiable:"true"` // beta filter on the Linear Regression of trendLine
TrailingCallbackRate []float64 `json:"trailingCallbackRate" modifiable:"true"`
TrailingActivationRatio []float64 `json:"trailingActivationRatio" modifiable:"true"`
buyPrice float64 `persistence:"buy_price"`
sellPrice float64 `persistence:"sell_price"`
highestPrice float64 `persistence:"highest_price"`
lowestPrice float64 `persistence:"lowest_price"`
buyPrice float64 `persistence:"buy_price"` // price when a long position is opened
sellPrice float64 `persistence:"sell_price"` // price when a short position is opened
highestPrice float64 `persistence:"highest_price"` // highestPrice when the position is opened
lowestPrice float64 `persistence:"lowest_price"` // lowestPrice when the position is opened
// This is not related to trade but for statistics graph generation
// Will deduct fee in percentage from every trade
GraphPNLDeductFee bool `json:"graphPNLDeductFee"`
GraphPNLPath string `json:"graphPNLPath"`
GraphCumPNLPath string `json:"graphCumPNLPath"`
// Whether to generate graph when shutdown
GenerateGraph bool `json:"generateGraph"`
CanvasPath string `json:"canvasPath"` // backtest related. the path to store the indicator graph
GraphPNLPath string `json:"graphPNLPath"` // backtest related. the path to store the pnl % graph per trade graph.
GraphCumPNLPath string `json:"graphCumPNLPath"` // backtest related. the path to store the asset changes in graph
GraphElapsedPath string `json:"graphElapsedPath"` // the path to store the elapsed time in ms
GenerateGraph bool `json:"generateGraph"` // whether to generate graph when shutdown
ExitMethods bbgo.ExitMethodSet `json:"exits"`
Session *bbgo.ExchangeSession
@ -135,15 +138,33 @@ func (s *Strategy) InstanceID() string {
func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {
// by default, bbgo only pre-subscribe 1000 klines.
// this is not enough if we're subscribing 30m intervals using SerialMarketDataStore
maxWindow := (s.Window + s.SmootherWindow + s.FisherTransformWindow) * s.Interval.Minutes()
bbgo.KLinePreloadLimit = int64((maxWindow/1000 + 1) * 1000)
log.Errorf("set kLinePreloadLimit to %d, %d %d", bbgo.KLinePreloadLimit, s.Interval.Minutes(), maxWindow)
session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
Interval: types.Interval1m,
})
if !bbgo.IsBackTesting {
session.Subscribe(types.BookTickerChannel, s.Symbol, types.SubscribeOptions{})
session.Subscribe(types.MarketTradeChannel, s.Symbol, types.SubscribeOptions{})
// able to preload
if s.MinInterval.Milliseconds() >= types.Interval1s.Milliseconds() && s.MinInterval.Milliseconds()%types.Interval1s.Milliseconds() == 0 {
maxWindow := (s.Window + s.SmootherWindow + s.FisherTransformWindow) * (s.Interval.Milliseconds() / s.MinInterval.Milliseconds())
bbgo.KLinePreloadLimit = int64((maxWindow/1000 + 1) * 1000)
session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
Interval: s.MinInterval,
})
} else {
bbgo.KLinePreloadLimit = 0
}
} else {
maxWindow := (s.Window + s.SmootherWindow + s.FisherTransformWindow) * (s.Interval.Milliseconds() / s.MinInterval.Milliseconds())
bbgo.KLinePreloadLimit = int64((maxWindow/1000 + 1) * 1000)
// gave up preload
if s.Interval.Milliseconds() < s.MinInterval.Milliseconds() {
bbgo.KLinePreloadLimit = 0
}
log.Errorf("set kLinePreloadLimit to %d, %d %d", bbgo.KLinePreloadLimit, s.Interval.Milliseconds()/s.MinInterval.Milliseconds(), maxWindow)
if bbgo.KLinePreloadLimit > 0 {
session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
Interval: s.MinInterval,
})
}
}
s.ExitMethods.SetAndSubscribe(session, s)
}
@ -208,6 +229,9 @@ func (s *Strategy) initIndicators(store *bbgo.SerialMarketDataStore) error {
s.atr = &indicator.ATR{IntervalWindow: types.IntervalWindow{Interval: s.Interval, Window: s.ATRWindow}}
s.trendLine = &indicator.EWMA{IntervalWindow: types.IntervalWindow{Interval: s.Interval, Window: s.TrendWindow}}
if bbgo.KLinePreloadLimit == 0 {
return nil
}
klines, ok := store.KLinesOfInterval(s.Interval)
klinesLength := len(*klines)
if !ok || klinesLength == 0 {
@ -229,20 +253,19 @@ func (s *Strategy) initIndicators(store *bbgo.SerialMarketDataStore) error {
if s.frameKLine != nil && klines != nil {
s.frameKLine.Set(&(*klines)[len(*klines)-1])
}
klines, ok = store.KLinesOfInterval(types.Interval1m)
klines, ok = store.KLinesOfInterval(s.MinInterval)
klinesLength = len(*klines)
if !ok || klinesLength == 0 {
return errors.New("klines not exists")
}
log.Infof("loaded %d klines1m", klinesLength)
if s.kline1m != nil && klines != nil {
s.kline1m.Set(&(*klines)[len(*klines)-1])
log.Infof("loaded %d klines%s", klinesLength, s.MinInterval)
if s.klineMin != nil && klines != nil {
s.klineMin.Set(&(*klines)[len(*klines)-1])
}
s.startTime = s.kline1m.StartTime.Time().Add(s.kline1m.Interval.Duration())
return nil
}
func (s *Strategy) smartCancel(ctx context.Context, pricef, atr float64) (int, error) {
func (s *Strategy) smartCancel(ctx context.Context, pricef, atr float64, syscounter int) (int, error) {
nonTraded := s.GeneralOrderExecutor.ActiveMakerOrders().Orders()
if len(nonTraded) > 0 {
if len(nonTraded) > 1 {
@ -254,17 +277,21 @@ func (s *Strategy) smartCancel(ctx context.Context, pricef, atr float64) (int, e
if order.Status != types.OrderStatusNew && order.Status != types.OrderStatusPartiallyFilled {
continue
}
log.Warnf("%v | counter: %d, system: %d", order, s.orderPendingCounter[order.OrderID], s.minutesCounter)
if s.minutesCounter-s.orderPendingCounter[order.OrderID] > s.PendingMinutes {
s.pendingLock.Lock()
counter := s.orderPendingCounter[order.OrderID]
s.pendingLock.Unlock()
log.Warnf("%v | counter: %d, system: %d", order, counter, syscounter)
if syscounter-counter > s.PendingMinInterval {
toCancel = true
} else if order.Side == types.SideTypeBuy {
// 75% of the probability
if order.Price.Float64()+s.stdevHigh.Last()*2 <= pricef {
if order.Price.Float64()+atr*2 <= pricef {
toCancel = true
}
} else if order.Side == types.SideTypeSell {
// 75% of the probability
if order.Price.Float64()-s.stdevLow.Last()*2 >= pricef {
if order.Price.Float64()-atr*2 >= pricef {
toCancel = true
}
} else {
@ -272,10 +299,22 @@ func (s *Strategy) smartCancel(ctx context.Context, pricef, atr float64) (int, e
}
}
if toCancel {
err := s.GeneralOrderExecutor.GracefulCancel(ctx)
err := s.GeneralOrderExecutor.CancelNoWait(ctx)
// TODO: clean orderPendingCounter on cancel/trade
for _, order := range nonTraded {
s.pendingLock.Lock()
counter := s.orderPendingCounter[order.OrderID]
delete(s.orderPendingCounter, order.OrderID)
s.pendingLock.Unlock()
if order.Side == types.SideTypeSell {
if s.maxCounterSellCanceled < counter {
s.maxCounterSellCanceled = counter
}
} else {
if s.maxCounterBuyCanceled < counter {
s.maxCounterBuyCanceled = counter
}
}
}
log.Warnf("cancel all %v", err)
return 0, err
@ -325,7 +364,6 @@ func (s *Strategy) initTickerFunctions(ctx context.Context) {
bestBid := ticker.Buy
bestAsk := ticker.Sell
var pricef float64
if !util.TryLock(&s.lock) {
return
}
@ -336,37 +374,10 @@ func (s *Strategy) initTickerFunctions(ctx context.Context) {
} else {
s.midPrice = bestBid
}
pricef = s.midPrice.Float64()
s.lock.Unlock()
if !util.TryLock(&s.positionLock) {
return
}
// we removed realtime stoploss and trailingStop.
if s.highestPrice > 0 && s.highestPrice < pricef {
s.highestPrice = pricef
}
if s.lowestPrice > 0 && s.lowestPrice > pricef {
s.lowestPrice = pricef
}
if s.CheckStopLoss() {
s.positionLock.Unlock()
s.ClosePosition(ctx, fixedpoint.One)
return
}
// for trailing stoploss during the realtime
if s.NoTrailingStopLoss || s.TrailingStopLossType == "kline" {
s.positionLock.Unlock()
return
}
exitCondition := s.trailingCheck(pricef, "short") || s.trailingCheck(pricef, "long")
s.positionLock.Unlock()
if exitCondition {
s.ClosePosition(ctx, fixedpoint.One)
}
})
s.getLastPrice = func() (lastPrice fixedpoint.Value) {
var ok bool
@ -387,102 +398,6 @@ func (s *Strategy) initTickerFunctions(ctx context.Context) {
}
func (s *Strategy) DrawIndicators(time types.Time) *types.Canvas {
canvas := types.NewCanvas(s.InstanceID(), s.Interval)
Length := s.priceLines.Length()
if Length > 300 {
Length = 300
}
log.Infof("draw indicators with %d data", Length)
mean := s.priceLines.Mean(Length)
highestPrice := s.priceLines.Minus(mean).Abs().Highest(Length)
highestDrift := s.drift.Abs().Highest(Length)
hi := s.drift.drift.Abs().Highest(Length)
ratio := highestPrice / highestDrift
// canvas.Plot("upband", s.ma.Add(s.stdevHigh), time, Length)
canvas.Plot("ma", s.ma, time, Length)
// canvas.Plot("downband", s.ma.Minus(s.stdevLow), time, Length)
fmt.Printf("%f %f\n", highestPrice, hi)
canvas.Plot("trend", s.trendLine, time, Length)
canvas.Plot("drift", s.drift.Mul(ratio).Add(mean), time, Length)
canvas.Plot("driftOrig", s.drift.drift.Mul(highestPrice/hi).Add(mean), time, Length)
canvas.Plot("zero", types.NumberSeries(mean), time, Length)
canvas.Plot("price", s.priceLines, time, Length)
return canvas
}
func (s *Strategy) DrawPNL(profit types.Series) *types.Canvas {
canvas := types.NewCanvas(s.InstanceID())
log.Errorf("pnl Highest: %f, Lowest: %f", types.Highest(profit, profit.Length()), types.Lowest(profit, profit.Length()))
length := profit.Length()
if s.GraphPNLDeductFee {
canvas.PlotRaw("pnl % (with Fee Deducted)", profit, length)
} else {
canvas.PlotRaw("pnl %", profit, length)
}
canvas.YAxis = chart.YAxis{
ValueFormatter: func(v interface{}) string {
if vf, isFloat := v.(float64); isFloat {
return fmt.Sprintf("%.4f", vf)
}
return ""
},
}
canvas.PlotRaw("1", types.NumberSeries(1), length)
return canvas
}
func (s *Strategy) DrawCumPNL(cumProfit types.Series) *types.Canvas {
canvas := types.NewCanvas(s.InstanceID())
canvas.PlotRaw("cummulative pnl", cumProfit, cumProfit.Length())
canvas.YAxis = chart.YAxis{
ValueFormatter: func(v interface{}) string {
if vf, isFloat := v.(float64); isFloat {
return fmt.Sprintf("%.4f", vf)
}
return ""
},
}
return canvas
}
func (s *Strategy) Draw(time types.Time, profit types.Series, cumProfit types.Series) {
canvas := s.DrawIndicators(time)
f, err := os.Create(s.CanvasPath)
if err != nil {
log.WithError(err).Errorf("cannot create on %s", s.CanvasPath)
return
}
defer f.Close()
if err := canvas.Render(chart.PNG, f); err != nil {
log.WithError(err).Errorf("cannot render in drift")
}
canvas = s.DrawPNL(profit)
f, err = os.Create(s.GraphPNLPath)
if err != nil {
log.WithError(err).Errorf("open pnl")
return
}
defer f.Close()
if err := canvas.Render(chart.PNG, f); err != nil {
log.WithError(err).Errorf("render pnl")
}
canvas = s.DrawCumPNL(cumProfit)
f, err = os.Create(s.GraphCumPNLPath)
if err != nil {
log.WithError(err).Errorf("open cumpnl")
return
}
defer f.Close()
if err := canvas.Render(chart.PNG, f); err != nil {
log.WithError(err).Errorf("render cumpnl")
}
}
// Sending new rebalance orders cost too much.
// Modify the position instead to expect the strategy itself rebalance on Close
func (s *Strategy) Rebalance(ctx context.Context) {
@ -545,8 +460,8 @@ func (s *Strategy) CalcAssetValue(price fixedpoint.Value) fixedpoint.Value {
return balances[s.Market.BaseCurrency].Total().Mul(price).Add(balances[s.Market.QuoteCurrency].Total())
}
func (s *Strategy) klineHandler1m(ctx context.Context, kline types.KLine) {
s.kline1m.Set(&kline)
func (s *Strategy) klineHandlerMin(ctx context.Context, kline types.KLine, counter int) {
s.klineMin.Set(&kline)
if s.Status != types.StrategyStatusRunning {
return
}
@ -564,58 +479,59 @@ func (s *Strategy) klineHandler1m(ctx context.Context, kline types.KLine) {
if s.highestPrice > 0 && highf > s.highestPrice {
s.highestPrice = highf
}
s.positionLock.Unlock()
numPending := 0
var err error
if numPending, err = s.smartCancel(ctx, pricef, atr); err != nil {
if numPending, err = s.smartCancel(ctx, pricef, atr, counter); err != nil {
log.WithError(err).Errorf("cannot cancel orders")
s.positionLock.Unlock()
return
}
if numPending > 0 {
s.positionLock.Unlock()
return
}
if s.NoTrailingStopLoss || s.TrailingStopLossType == "realtime" {
s.positionLock.Unlock()
if s.NoTrailingStopLoss {
return
}
exitCondition := s.CheckStopLoss() || s.trailingCheck(highf, "short") || s.trailingCheck(lowf, "long")
s.positionLock.Unlock()
if exitCondition {
_ = s.ClosePosition(ctx, fixedpoint.One)
}
}
func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
var driftPred, atr float64
var drift []float64
func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine, counter int) {
start := time.Now()
defer func() {
end := time.Now()
elapsed := end.Sub(start)
s.elapsed.Update(float64(elapsed) / 1000000)
}()
s.frameKLine.Set(&kline)
source := s.GetSource(&kline)
sourcef := source.Float64()
s.priceLines.Update(sourcef)
s.ma.Update(sourcef)
//s.ma.Update(sourcef)
s.trendLine.Update(sourcef)
s.drift.Update(sourcef, kline.Volume.Abs().Float64())
s.atr.PushK(kline)
atr := s.atr.Last()
driftPred = s.drift.Predict(s.PredictOffset)
ddriftPred := s.drift.drift.Predict(s.PredictOffset)
atr = s.atr.Last()
price := s.getLastPrice()
price := kline.Close //s.getLastPrice()
pricef := price.Float64()
lowf := math.Min(kline.Low.Float64(), pricef)
highf := math.Max(kline.High.Float64(), pricef)
lowdiff := s.ma.Last() - lowf
lowdiff := pricef - lowf
s.stdevLow.Update(lowdiff)
highdiff := highf - s.ma.Last()
highdiff := highf - pricef
s.stdevHigh.Update(highdiff)
drift = s.drift.Array(2)
drift := s.drift.Array(2)
if len(drift) < 2 || len(drift) < s.PredictOffset {
return
}
@ -628,79 +544,81 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
return
}
log.Infof("highdiff: %3.2f open: %8v, close: %8v, high: %8v, low: %8v, time: %v %v", s.stdevHigh.Last(), kline.Open, kline.Close, kline.High, kline.Low, kline.StartTime, kline.EndTime)
s.positionLock.Lock()
log.Infof("highdiff: %3.2f ma: %.2f, open: %8v, close: %8v, high: %8v, low: %8v, time: %v %v", s.stdevHigh.Last(), s.ma.Last(), kline.Open, kline.Close, kline.High, kline.Low, kline.StartTime, kline.EndTime)
if s.lowestPrice > 0 && lowf < s.lowestPrice {
s.lowestPrice = lowf
}
if s.highestPrice > 0 && highf > s.highestPrice {
s.highestPrice = highf
}
s.positionLock.Unlock()
if !s.NoRebalance {
s.Rebalance(ctx)
}
balances := s.GeneralOrderExecutor.Session().GetAccount().Balances()
bbgo.Notify("source: %.4f, price: %.4f, driftPred: %.4f, ddriftPred: %.4f, drift[1]: %.4f, ddrift[1]: %.4f, atr: %.4f, lowf %.4f, highf: %.4f lowest: %.4f highest: %.4f sp %.4f bp %.4f",
sourcef, pricef, driftPred, ddriftPred, drift[1], ddrift[1], atr, lowf, highf, s.lowestPrice, s.highestPrice, s.sellPrice, s.buyPrice)
// Notify will parse args to strings and process separately
bbgo.Notify("balances: [Total] %v %s [Base] %s(%v %s) [Quote] %s",
s.CalcAssetValue(price),
s.Market.QuoteCurrency,
balances[s.Market.BaseCurrency].String(),
balances[s.Market.BaseCurrency].Total().Mul(price),
s.Market.QuoteCurrency,
balances[s.Market.QuoteCurrency].String(),
)
if s.Debug {
balances := s.GeneralOrderExecutor.Session().GetAccount().Balances()
bbgo.Notify("source: %.4f, price: %.4f, drift[0]: %.4f, ddrift[0]: %.4f, lowf %.4f, highf: %.4f lowest: %.4f highest: %.4f sp %.4f bp %.4f",
sourcef, pricef, drift[0], ddrift[0], atr, lowf, highf, s.lowestPrice, s.highestPrice, s.sellPrice, s.buyPrice)
// Notify will parse args to strings and process separately
bbgo.Notify("balances: [Total] %v %s [Base] %s(%v %s) [Quote] %s",
s.CalcAssetValue(price),
s.Market.QuoteCurrency,
balances[s.Market.BaseCurrency].String(),
balances[s.Market.BaseCurrency].Total().Mul(price),
s.Market.QuoteCurrency,
balances[s.Market.QuoteCurrency].String(),
)
}
shortCondition := drift[1] >= 0 && drift[0] <= 0 || (drift[1] >= drift[0] && drift[1] <= 0) || ddrift[1] >= 0 && ddrift[0] <= 0 || (ddrift[1] >= ddrift[0] && ddrift[1] <= 0)
longCondition := drift[1] <= 0 && drift[0] >= 0 || (drift[1] <= drift[0] && drift[1] >= 0) || ddrift[1] <= 0 && ddrift[0] >= 0 || (ddrift[1] <= ddrift[0] && ddrift[1] >= 0)
if shortCondition && longCondition {
if drift[1] > drift[0] {
if s.priceLines.Index(1) > s.priceLines.Last() {
longCondition = false
} else {
shortCondition = false
}
}
exitCondition := s.CheckStopLoss() || s.trailingCheck(pricef, "short") || s.trailingCheck(pricef, "long")
exitCondition := !s.NoTrailingStopLoss && (s.CheckStopLoss() || s.trailingCheck(pricef, "short") || s.trailingCheck(pricef, "long"))
if exitCondition {
s.positionLock.Unlock()
if err := s.GeneralOrderExecutor.GracefulCancel(ctx); err != nil {
if exitCondition || longCondition || shortCondition {
var err error
var hold int
if hold, err = s.smartCancel(ctx, pricef, atr, counter); err != nil {
log.WithError(err).Errorf("cannot cancel orders")
}
if hold > 0 {
return
}
_ = s.ClosePosition(ctx, fixedpoint.One)
if shortCondition || longCondition {
s.positionLock.Lock()
} else {
return
} else {
if _, err := s.smartCancel(ctx, pricef, atr, counter); err != nil {
log.WithError(err).Errorf("cannot cancel orders")
}
return
}
if longCondition {
if err := s.GeneralOrderExecutor.GracefulCancel(ctx); err != nil {
log.WithError(err).Errorf("cannot cancel orders")
s.positionLock.Unlock()
return
}
source = source.Sub(fixedpoint.NewFromFloat(s.stdevLow.Last() * s.HighLowVarianceMultiplier))
if source.Compare(price) > 0 {
source = price
}
/*source = fixedpoint.NewFromFloat(s.ma.Last() - s.stdevLow.Last()*s.HighLowVarianceMultiplier)
if source.Compare(price) > 0 {
source = price
}
sourcef = source.Float64()*/
log.Infof("source in long %v %v %f", source, price, s.stdevLow.Last())
s.positionLock.Unlock()
opt := s.OpenPositionOptions
opt.Long = true
opt.LimitOrder = true
// force to use market taker
if counter-s.maxCounterBuyCanceled <= s.PendingMinInterval {
opt.LimitOrder = false
}
opt.Price = source
opt.Tags = []string{"long"}
createdOrders, err := s.GeneralOrderExecutor.OpenPosition(ctx, opt)
if err != nil {
errs := filterErrors(multierr.Errors(err))
@ -710,35 +628,34 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
}
return
}
log.Infof("orders %v", createdOrders)
if createdOrders != nil {
s.orderPendingCounter[createdOrders[0].OrderID] = s.minutesCounter
for _, o := range createdOrders {
if o.Status == types.OrderStatusNew || o.Status == types.OrderStatusPartiallyFilled {
s.pendingLock.Lock()
s.orderPendingCounter[o.OrderID] = counter
s.pendingLock.Unlock()
}
}
}
return
}
if shortCondition {
if err := s.GeneralOrderExecutor.GracefulCancel(ctx); err != nil {
log.WithError(err).Errorf("cannot cancel orders")
s.positionLock.Unlock()
return
}
source = source.Add(fixedpoint.NewFromFloat(s.stdevHigh.Last() * s.HighLowVarianceMultiplier))
if source.Compare(price) < 0 {
source = price
}
/*source = fixedpoint.NewFromFloat(s.ma.Last() + s.stdevHigh.Last()*s.HighLowVarianceMultiplier)
if source.Compare(price) < 0 {
source = price
}
sourcef = source.Float64()*/
log.Infof("source in short: %v", source)
s.positionLock.Unlock()
opt := s.OpenPositionOptions
opt.Short = true
opt.Price = source
opt.LimitOrder = true
if counter-s.maxCounterSellCanceled <= s.PendingMinInterval {
opt.LimitOrder = false
}
opt.Tags = []string{"short"}
createdOrders, err := s.GeneralOrderExecutor.OpenPosition(ctx, opt)
if err != nil {
@ -750,11 +667,16 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
}
log.Infof("orders %v", createdOrders)
if createdOrders != nil {
s.orderPendingCounter[createdOrders[0].OrderID] = s.minutesCounter
for _, o := range createdOrders {
if o.Status == types.OrderStatusNew || o.Status == types.OrderStatusPartiallyFilled {
s.pendingLock.Lock()
s.orderPendingCounter[o.OrderID] = counter
s.pendingLock.Unlock()
}
}
}
return
}
s.positionLock.Unlock()
}
func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, session *bbgo.ExchangeSession) error {
@ -800,7 +722,6 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
s.GeneralOrderExecutor.Bind()
s.orderPendingCounter = make(map[uint64]int)
s.minutesCounter = 0
// Exit methods from config
for _, method := range s.ExitMethods {
@ -821,13 +742,10 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
}
s.GeneralOrderExecutor.TradeCollector().OnTrade(func(trade types.Trade, _profit, _netProfit fixedpoint.Value) {
s.p.AddTrade(trade)
order, ok := s.GeneralOrderExecutor.TradeCollector().OrderStore().Get(trade.OrderID)
if !ok {
panic(fmt.Sprintf("cannot find order: %v", trade))
}
tag := order.Tag
price := trade.Price.Float64()
s.pendingLock.Lock()
delete(s.orderPendingCounter, trade.OrderID)
s.pendingLock.Unlock()
if s.buyPrice > 0 {
profit.Update(modify(price / s.buyPrice))
@ -837,19 +755,18 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
cumProfit.Update(s.CalcAssetValue(trade.Price).Float64())
}
s.positionLock.Lock()
defer s.positionLock.Unlock()
if s.p.IsDust(trade.Price) {
s.buyPrice = 0
s.sellPrice = 0
s.highestPrice = 0
s.lowestPrice = 0
} else if s.p.IsLong() {
s.buyPrice = s.p.ApproximateAverageCost.Float64() // trade.Price.Float64()
s.buyPrice = s.p.ApproximateAverageCost.Float64()
s.sellPrice = 0
s.highestPrice = math.Max(s.buyPrice, s.highestPrice)
s.lowestPrice = s.buyPrice
} else if s.p.IsShort() {
s.sellPrice = s.p.ApproximateAverageCost.Float64() // trade.Price.Float64()
s.sellPrice = s.p.ApproximateAverageCost.Float64()
s.buyPrice = 0
s.highestPrice = s.sellPrice
if s.lowestPrice == 0 {
@ -858,55 +775,20 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
s.lowestPrice = math.Min(s.lowestPrice, s.sellPrice)
}
}
bbgo.Notify("tag: %s, sp: %.4f bp: %.4f hp: %.4f lp: %.4f, trade: %s, pos: %s", tag, s.sellPrice, s.buyPrice, s.highestPrice, s.lowestPrice, trade.String(), s.p.String())
s.positionLock.Unlock()
})
s.frameKLine = &types.KLine{}
s.kline1m = &types.KLine{}
s.klineMin = &types.KLine{}
s.priceLines = types.NewQueue(300)
s.elapsed = types.NewQueue(60000)
s.initTickerFunctions(ctx)
startTime := s.Environment.StartTime()
s.TradeStats.SetIntervalProfitCollector(types.NewIntervalProfitCollector(types.Interval1d, startTime))
s.TradeStats.SetIntervalProfitCollector(types.NewIntervalProfitCollector(types.Interval1w, startTime))
s.startTime = s.Environment.StartTime()
s.TradeStats.SetIntervalProfitCollector(types.NewIntervalProfitCollector(types.Interval1d, s.startTime))
s.TradeStats.SetIntervalProfitCollector(types.NewIntervalProfitCollector(types.Interval1w, s.startTime))
// default value: use 1m kline
if !s.NoTrailingStopLoss && s.IsBackTesting() || s.TrailingStopLossType == "" {
s.TrailingStopLossType = "kline"
}
bbgo.RegisterCommand("/draw", "Draw Indicators", func(reply interact.Reply) {
canvas := s.DrawIndicators(s.frameKLine.StartTime)
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render indicators in drift")
reply.Message(fmt.Sprintf("[error] cannot render indicators in drift: %v", err))
return
}
bbgo.SendPhoto(&buffer)
})
bbgo.RegisterCommand("/pnl", "Draw PNL(%) per trade", func(reply interact.Reply) {
canvas := s.DrawPNL(&profit)
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render pnl in drift")
reply.Message(fmt.Sprintf("[error] cannot render pnl in drift: %v", err))
return
}
bbgo.SendPhoto(&buffer)
})
bbgo.RegisterCommand("/cumpnl", "Draw Cummulative PNL(Quote)", func(reply interact.Reply) {
canvas := s.DrawCumPNL(&cumProfit)
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render cumpnl in drift")
reply.Message(fmt.Sprintf("[error] canot render cumpnl in drift: %v", err))
return
}
bbgo.SendPhoto(&buffer)
})
s.InitDrawCommands(&profit, &cumProfit)
bbgo.RegisterCommand("/config", "Show latest config", func(reply interact.Reply) {
var buffer bytes.Buffer
@ -933,21 +815,23 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
bbgo.RegisterModifier(s)
// event trigger order: s.Interval => Interval1m
store, ok := session.SerialMarketDataStore(s.Symbol, []types.Interval{s.Interval, types.Interval1m})
// event trigger order: s.Interval => s.MinInterval
store, ok := session.SerialMarketDataStore(ctx, s.Symbol, []types.Interval{s.Interval, s.MinInterval}, !bbgo.IsBackTesting)
if !ok {
panic("cannot get 1m history")
panic("cannot get " + s.MinInterval + " history")
}
if err := s.initIndicators(store); err != nil {
log.WithError(err).Errorf("initIndicator failed")
return nil
}
//var lastK types.KLine
store.OnKLineClosed(func(kline types.KLine) {
s.minutesCounter = int(kline.StartTime.Time().Add(kline.Interval.Duration()).Sub(s.startTime).Minutes())
counter := int(kline.StartTime.Time().Add(kline.Interval.Duration()).Sub(s.startTime).Milliseconds()) / s.MinInterval.Milliseconds()
if kline.Interval == s.Interval {
s.klineHandler(ctx, kline)
} else if kline.Interval == types.Interval1m {
s.klineHandler1m(ctx, kline)
s.klineHandler(ctx, kline, counter)
} else if kline.Interval == s.MinInterval {
s.klineHandlerMin(ctx, kline, counter)
}
})

71
pkg/strategy/elliottwave/draw.go
View File

@ -13,47 +13,46 @@ import (
func (s *Strategy) InitDrawCommands(store *bbgo.SerialMarketDataStore, profit, cumProfit types.Series) {
bbgo.RegisterCommand("/draw", "Draw Indicators", func(reply interact.Reply) {
canvas := s.DrawIndicators(store)
if canvas == nil {
reply.Message("cannot render indicators")
return
}
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render indicators in ewo")
reply.Message(fmt.Sprintf("[error] cannot render indicators in ewo: %v", err))
return
}
bbgo.SendPhoto(&buffer)
go func() {
canvas := s.DrawIndicators(store)
if canvas == nil {
reply.Send("cannot render indicators")
return
}
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render indicators in ewo")
return
}
bbgo.SendPhoto(&buffer)
}()
})
bbgo.RegisterCommand("/pnl", "Draw PNL(%) per trade", func(reply interact.Reply) {
canvas := s.DrawPNL(profit)
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render pnl in drift")
reply.Message(fmt.Sprintf("[error] cannot render pnl in ewo: %v", err))
return
}
bbgo.SendPhoto(&buffer)
go func() {
canvas := s.DrawPNL(profit)
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render pnl in ewo")
return
}
bbgo.SendPhoto(&buffer)
}()
})
bbgo.RegisterCommand("/cumpnl", "Draw Cummulative PNL(Quote)", func(reply interact.Reply) {
canvas := s.DrawCumPNL(cumProfit)
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render cumpnl in drift")
reply.Message(fmt.Sprintf("[error] canot render cumpnl in drift: %v", err))
return
}
bbgo.SendPhoto(&buffer)
go func() {
canvas := s.DrawCumPNL(cumProfit)
var buffer bytes.Buffer
if err := canvas.Render(chart.PNG, &buffer); err != nil {
log.WithError(err).Errorf("cannot render cumpnl in ewo")
return
}
bbgo.SendPhoto(&buffer)
}()
})
}
func (s *Strategy) DrawIndicators(store *bbgo.SerialMarketDataStore) *types.Canvas {
klines, ok := store.KLinesOfInterval(types.Interval1m)
if !ok {
return nil
}
time := (*klines)[len(*klines)-1].StartTime
time := types.Time(s.startTime)
canvas := types.NewCanvas(s.InstanceID(), s.Interval)
Length := s.priceLines.Length()
if Length > 300 {
@ -109,10 +108,10 @@ func (s *Strategy) Draw(store *bbgo.SerialMarketDataStore, profit, cumProfit typ
log.WithError(err).Errorf("cannot create on path " + s.GraphIndicatorPath)
return
}
defer f.Close()
if err = canvas.Render(chart.PNG, f); err != nil {
log.WithError(err).Errorf("cannot render elliottwave")
}
f.Close()
canvas = s.DrawPNL(profit)
f, err = os.Create(s.GraphPNLPath)
@ -120,19 +119,19 @@ func (s *Strategy) Draw(store *bbgo.SerialMarketDataStore, profit, cumProfit typ
log.WithError(err).Errorf("cannot create on path " + s.GraphPNLPath)
return
}
defer f.Close()
if err = canvas.Render(chart.PNG, f); err != nil {
log.WithError(err).Errorf("cannot render pnl")
return
}
f.Close()
canvas = s.DrawCumPNL(cumProfit)
f, err = os.Create(s.GraphCumPNLPath)
if err != nil {
log.WithError(err).Errorf("cannot create on path " + s.GraphCumPNLPath)
return
}
defer f.Close()
if err = canvas.Render(chart.PNG, f); err != nil {
log.WithError(err).Errorf("cannot render cumpnl")
}
f.Close()
}

126
pkg/strategy/elliottwave/strategy.go
View File

@ -43,13 +43,14 @@ type Strategy struct {
types.Market
Session *bbgo.ExchangeSession
Interval types.Interval `json:"interval"`
Stoploss fixedpoint.Value `json:"stoploss" modifiable:"true"`
WindowATR int `json:"windowATR"`
WindowQuick int `json:"windowQuick"`
WindowSlow int `json:"windowSlow"`
PendingMinutes int `json:"pendingMinutes" modifiable:"true"`
UseHeikinAshi bool `json:"useHeikinAshi"`
Interval types.Interval `json:"interval"`
MinInterval types.Interval `json:"minInterval"`
Stoploss fixedpoint.Value `json:"stoploss" modifiable:"true"`
WindowATR int `json:"windowATR"`
WindowQuick int `json:"windowQuick"`
WindowSlow int `json:"windowSlow"`
PendingMinInterval int `json:"pendingMinInterval" modifiable:"true"`
UseHeikinAshi bool `json:"useHeikinAshi"`
// whether to draw graph or not by the end of backtest
DrawGraph bool `json:"drawGraph"`
@ -74,7 +75,7 @@ type Strategy struct {
// for smart cancel
orderPendingCounter map[uint64]int
startTime time.Time
minutesCounter int
counter int
// for position
buyPrice float64 `persistence:"buy_price"`
@ -101,12 +102,22 @@ func (s *Strategy) InstanceID() string {
func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {
// by default, bbgo only pre-subscribe 1000 klines.
// this is not enough if we're subscribing 30m intervals using SerialMarketDataStore
bbgo.KLinePreloadLimit = int64((s.Interval.Minutes()*s.WindowSlow/1000 + 1) + 1000)
session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
Interval: types.Interval1m,
})
if !bbgo.IsBackTesting {
session.Subscribe(types.BookTickerChannel, s.Symbol, types.SubscribeOptions{})
session.Subscribe(types.MarketTradeChannel, s.Symbol, types.SubscribeOptions{})
if s.MinInterval.Milliseconds() >= types.Interval1s.Milliseconds() && s.MinInterval.Milliseconds()%types.Interval1s.Milliseconds() == 0 {
bbgo.KLinePreloadLimit = int64(((s.Interval.Milliseconds()/s.MinInterval.Milliseconds())*s.WindowSlow/1000 + 1) + 1000)
session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
Interval: s.MinInterval,
})
} else {
bbgo.KLinePreloadLimit = 0
}
} else {
bbgo.KLinePreloadLimit = int64((s.Interval.Milliseconds()/s.MinInterval.Milliseconds()*s.WindowSlow/1000 + 1) + 1000)
session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{
Interval: s.MinInterval,
})
}
s.ExitMethods.SetAndSubscribe(session, s)
}
@ -162,8 +173,6 @@ func (s *Strategy) initIndicators(store *bbgo.SerialMarketDataStore) error {
if !ok || klineLength == 0 {
return errors.New("klines not exists")
}
tmpK := (*klines)[klineLength-1]
s.startTime = tmpK.StartTime.Time().Add(tmpK.Interval.Duration())
s.heikinAshi = NewHeikinAshi(500)
for _, kline := range *klines {
@ -185,9 +194,9 @@ func (s *Strategy) smartCancel(ctx context.Context, pricef float64) int {
if order.Status != types.OrderStatusNew && order.Status != types.OrderStatusPartiallyFilled {
continue
}
log.Warnf("%v | counter: %d, system: %d", order, s.orderPendingCounter[order.OrderID], s.minutesCounter)
log.Warnf("%v | counter: %d, system: %d", order, s.orderPendingCounter[order.OrderID], s.counter)
toCancel := false
if s.minutesCounter-s.orderPendingCounter[order.OrderID] >= s.PendingMinutes {
if s.counter-s.orderPendingCounter[order.OrderID] >= s.PendingMinInterval {
toCancel = true
} else if order.Side == types.SideTypeBuy {
if order.Price.Float64()+s.atr.Last()*2 <= pricef {
@ -202,7 +211,7 @@ func (s *Strategy) smartCancel(ctx context.Context, pricef float64) int {
panic("not supported side for the order")
}
if toCancel {
err := s.GeneralOrderExecutor.GracefulCancel(ctx, order)
err := s.GeneralOrderExecutor.CancelNoWait(ctx, order)
if err == nil {
delete(s.orderPendingCounter, order.OrderID)
} else {
@ -226,6 +235,9 @@ func (s *Strategy) trailingCheck(price float64, direction string) bool {
s.lowestPrice = price
}
isShort := direction == "short"
if isShort && s.sellPrice == 0 || !isShort && s.buyPrice == 0 {
return false
}
for i := len(s.TrailingCallbackRate) - 1; i >= 0; i-- {
trailingCallbackRate := s.TrailingCallbackRate[i]
trailingActivationRatio := s.TrailingActivationRatio[i]
@ -235,7 +247,7 @@ func (s *Strategy) trailingCheck(price float64, direction string) bool {
}
} else {
if (s.highestPrice-s.buyPrice)/s.buyPrice > trailingActivationRatio {
return (s.highestPrice-price)/price > trailingCallbackRate
return (s.highestPrice-price)/s.buyPrice > trailingCallbackRate
}
}
}
@ -315,7 +327,7 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
s.GeneralOrderExecutor.Bind()
s.orderPendingCounter = make(map[uint64]int)
s.minutesCounter = 0
s.counter = 0
for _, method := range s.ExitMethods {
method.Bind(session, s.GeneralOrderExecutor)
@ -342,29 +354,33 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
s.highestPrice = 0
s.lowestPrice = 0
} else if s.Position.IsLong() {
s.buyPrice = price
s.buyPrice = s.Position.ApproximateAverageCost.Float64()
s.sellPrice = 0
s.highestPrice = s.buyPrice
s.highestPrice = math.Max(s.buyPrice, s.highestPrice)
s.lowestPrice = 0
} else {
s.sellPrice = price
s.sellPrice = s.Position.ApproximateAverageCost.Float64()
s.buyPrice = 0
s.highestPrice = 0
s.lowestPrice = s.sellPrice
if s.lowestPrice == 0 {
s.lowestPrice = s.sellPrice
} else {
s.lowestPrice = math.Min(s.lowestPrice, s.sellPrice)
}
}
})
s.initTickerFunctions()
startTime := s.Environment.StartTime()
s.TradeStats.SetIntervalProfitCollector(types.NewIntervalProfitCollector(types.Interval1d, startTime))
s.TradeStats.SetIntervalProfitCollector(types.NewIntervalProfitCollector(types.Interval1w, startTime))
s.startTime = s.Environment.StartTime()
s.TradeStats.SetIntervalProfitCollector(types.NewIntervalProfitCollector(types.Interval1d, s.startTime))
s.TradeStats.SetIntervalProfitCollector(types.NewIntervalProfitCollector(types.Interval1w, s.startTime))
s.initOutputCommands()
// event trigger order: s.Interval => Interval1m
store, ok := session.SerialMarketDataStore(s.Symbol, []types.Interval{s.Interval, types.Interval1m})
// event trigger order: s.Interval => minInterval
store, ok := session.SerialMarketDataStore(ctx, s.Symbol, []types.Interval{s.Interval, s.MinInterval}, !bbgo.IsBackTesting)
if !ok {
panic("cannot get 1m history")
panic("cannot get " + s.MinInterval + " history")
}
if err := s.initIndicators(store); err != nil {
log.WithError(err).Errorf("initIndicator failed")
@ -372,11 +388,11 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
}
s.InitDrawCommands(store, &profit, &cumProfit)
store.OnKLineClosed(func(kline types.KLine) {
s.minutesCounter = int(kline.StartTime.Time().Add(kline.Interval.Duration()).Sub(s.startTime).Minutes())
s.counter = int(kline.StartTime.Time().Add(kline.Interval.Duration()).Sub(s.startTime).Milliseconds())
if kline.Interval == s.Interval {
s.klineHandler(ctx, kline)
} else if kline.Interval == types.Interval1m {
s.klineHandler1m(ctx, kline)
} else if kline.Interval == s.MinInterval {
s.klineHandlerMin(ctx, kline)
}
})
@ -401,7 +417,7 @@ func (s *Strategy) CalcAssetValue(price fixedpoint.Value) fixedpoint.Value {
return balances[s.Market.BaseCurrency].Total().Mul(price).Add(balances[s.Market.QuoteCurrency].Total())
}
func (s *Strategy) klineHandler1m(ctx context.Context, kline types.KLine) {
func (s *Strategy) klineHandlerMin(ctx context.Context, kline types.KLine) {
if s.Status != types.StrategyStatusRunning {
return
}
@ -468,15 +484,18 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
bull := kline.Close.Compare(kline.Open) > 0
balances := s.GeneralOrderExecutor.Session().GetAccount().Balances()
bbgo.Notify("source: %.4f, price: %.4f lowest: %.4f highest: %.4f sp %.4f bp %.4f", sourcef, pricef, s.lowestPrice, s.highestPrice, s.sellPrice, s.buyPrice)
bbgo.Notify("balances: [Total] %v %s [Base] %s(%v %s) [Quote] %s",
s.CalcAssetValue(price),
s.Market.QuoteCurrency,
balances[s.Market.BaseCurrency].String(),
balances[s.Market.BaseCurrency].Total().Mul(price),
s.Market.QuoteCurrency,
balances[s.Market.QuoteCurrency].String(),
)
startTime := kline.StartTime.Time()
if startTime.Round(time.Second) == startTime.Round(time.Minute) {
bbgo.Notify("source: %.4f, price: %.4f lowest: %.4f highest: %.4f sp %.4f bp %.4f", sourcef, pricef, s.lowestPrice, s.highestPrice, s.sellPrice, s.buyPrice)
bbgo.Notify("balances: [Total] %v %s [Base] %s(%v %s) [Quote] %s",
s.CalcAssetValue(price),
s.Market.QuoteCurrency,
balances[s.Market.BaseCurrency].String(),
balances[s.Market.BaseCurrency].Total().Mul(price),
s.Market.QuoteCurrency,
balances[s.Market.QuoteCurrency].String(),
)
}
shortCondition := ewo[0] < ewo[1] && ewo[1] >= ewo[2] && (ewo[1] <= ewo[2] || ewo[2] >= ewo[3]) || s.sellPrice == 0 && ewo[0] < ewo[1] && ewo[1] < ewo[2]
longCondition := ewo[0] > ewo[1] && ewo[1] <= ewo[2] && (ewo[1] >= ewo[2] || ewo[2] <= ewo[3]) || s.buyPrice == 0 && ewo[0] > ewo[1] && ewo[1] > ewo[2]
@ -484,18 +503,19 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
exitShortCondition := s.sellPrice > 0 && !shortCondition && s.sellPrice*(1.+stoploss) <= highf || s.sellPrice+atr <= highf || s.trailingCheck(highf, "short")
exitLongCondition := s.buyPrice > 0 && !longCondition && s.buyPrice*(1.-stoploss) >= lowf || s.buyPrice-atr >= lowf || s.trailingCheck(lowf, "long")
if exitShortCondition || exitLongCondition {
if err := s.GeneralOrderExecutor.GracefulCancel(ctx); err != nil {
log.WithError(err).Errorf("cannot cancel orders")
if exitShortCondition || exitLongCondition || (longCondition && bull) || (shortCondition && !bull) {
if hold := s.smartCancel(ctx, pricef); hold > 0 {
return
}
} else {
s.smartCancel(ctx, pricef)
return
}
if exitShortCondition || exitLongCondition {
s.ClosePosition(ctx, fixedpoint.One)
}
if longCondition && bull {
if err := s.GeneralOrderExecutor.GracefulCancel(ctx); err != nil {
log.WithError(err).Errorf("cannot cancel orders")
return
}
if source.Compare(price) > 0 {
source = price
}
@ -513,15 +533,11 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
return
}
if createdOrders != nil {
s.orderPendingCounter[createdOrders[0].OrderID] = s.minutesCounter
s.orderPendingCounter[createdOrders[0].OrderID] = s.counter
}
return
}
if shortCondition && !bull {
if err := s.GeneralOrderExecutor.GracefulCancel(ctx); err != nil {
log.WithError(err).Errorf("cannot cancel orders")
return
}
if source.Compare(price) < 0 {
source = price
}
@ -539,7 +555,7 @@ func (s *Strategy) klineHandler(ctx context.Context, kline types.KLine) {
return
}
if createdOrders != nil {
s.orderPendingCounter[createdOrders[0].OrderID] = s.minutesCounter
s.orderPendingCounter[createdOrders[0].OrderID] = s.counter
}
return
}

4
pkg/strategy/ewoDgtrd/strategy.go
View File

@ -145,7 +145,7 @@ func (inc *CCISTOCH) Update(cloze float64) {
func (inc *CCISTOCH) BuySignal() bool {
hasGrey := false
for i := 0; i < len(inc.ma.Values); i++ {
for i := 0; i < inc.ma.Values.Length(); i++ {
v := inc.ma.Index(i)
if v > inc.filterHigh {
return false
@ -161,7 +161,7 @@ func (inc *CCISTOCH) BuySignal() bool {
func (inc *CCISTOCH) SellSignal() bool {
hasGrey := false
for i := 0; i < len(inc.ma.Values); i++ {
for i := 0; i < inc.ma.Values.Length(); i++ {
v := inc.ma.Index(i)
if v < inc.filterLow {
return false

37
pkg/types/interval.go
View File

@ -25,8 +25,42 @@ func (i Interval) Seconds() int {
return m
}
// Milliseconds is specially handled, for better precision
// for ms level interval, calling Seconds and Minutes directly might trigger panic error
func (i Interval) Milliseconds() int {
t := 0
index := 0
for i, rn := range string(i) {
if rn >= '0' && rn <= '9' {
t = t*10 + int(rn-'0')
} else {
index = i
break
}
}
switch strings.ToLower(string(i[index:])) {
case "ms":
return t
case "s":
return t * 1000
case "m":
t *= 60
case "h":
t *= 60 * 60
case "d":
t *= 60 * 60 * 24
case "w":
t *= 60 * 60 * 24 * 7
case "mo":
t *= 60 * 60 * 24 * 30
default:
panic("unknown interval input: " + i)
}
return t * 1000
}
func (i Interval) Duration() time.Duration {
return time.Duration(i.Seconds()) * time.Second
return time.Duration(i.Milliseconds()) * time.Millisecond
}
func (i *Interval) UnmarshalJSON(b []byte) (err error) {
@ -53,6 +87,7 @@ func (s IntervalSlice) StringSlice() (slice []string) {
return slice
}
var Interval1ms = Interval("1ms")
var Interval1s = Interval("1s")
var Interval1m = Interval("1m")
var Interval3m = Interval("3m")