package harmonic import ( "context" "fmt" "os" "sync" "github.com/c9s/bbgo/pkg/bbgo" "github.com/c9s/bbgo/pkg/data/tsv" "github.com/c9s/bbgo/pkg/datatype/floats" "github.com/c9s/bbgo/pkg/fixedpoint" "github.com/c9s/bbgo/pkg/indicator" "github.com/c9s/bbgo/pkg/types" "github.com/sirupsen/logrus" ) const ID = "harmonic" var log = logrus.WithField("strategy", ID) func init() { bbgo.RegisterStrategy(ID, &Strategy{}) } type Strategy struct { Environment *bbgo.Environment Symbol string `json:"symbol"` Market types.Market types.IntervalWindow //bbgo.OpenPositionOptions // persistence fields Position *types.Position `persistence:"position"` ProfitStats *types.ProfitStats `persistence:"profit_stats"` TradeStats *types.TradeStats `persistence:"trade_stats"` ExitMethods bbgo.ExitMethodSet `json:"exits"` session *bbgo.ExchangeSession orderExecutor *bbgo.GeneralOrderExecutor bbgo.QuantityOrAmount // StrategyController bbgo.StrategyController shark *SHARK AccountValueCalculator *bbgo.AccountValueCalculator // whether to draw graph or not by the end of backtest DrawGraph bool `json:"drawGraph"` GraphPNLPath string `json:"graphPNLPath"` GraphCumPNLPath string `json:"graphCumPNLPath"` // for position buyPrice float64 `persistence:"buy_price"` sellPrice float64 `persistence:"sell_price"` highestPrice float64 `persistence:"highest_price"` lowestPrice float64 `persistence:"lowest_price"` // Accumulated profit report AccumulatedProfitReport *AccumulatedProfitReport `json:"accumulatedProfitReport"` } // AccumulatedProfitReport For accumulated profit report output type AccumulatedProfitReport struct { // AccumulatedProfitMAWindow Accumulated profit SMA window, in number of trades AccumulatedProfitMAWindow int `json:"accumulatedProfitMAWindow"` // IntervalWindow interval window, in days IntervalWindow int `json:"intervalWindow"` // NumberOfInterval How many intervals to output to TSV NumberOfInterval int `json:"NumberOfInterval"` // TsvReportPath The path to output report to TsvReportPath string `json:"tsvReportPath"` // AccumulatedDailyProfitWindow The window to sum up the daily profit, in days AccumulatedDailyProfitWindow int `json:"accumulatedDailyProfitWindow"` // Accumulated profit accumulatedProfit fixedpoint.Value accumulatedProfitPerDay floats.Slice previousAccumulatedProfit fixedpoint.Value // Accumulated profit MA accumulatedProfitMA *indicator.SMA accumulatedProfitMAPerDay floats.Slice // Daily profit dailyProfit floats.Slice // Accumulated fee accumulatedFee fixedpoint.Value accumulatedFeePerDay floats.Slice // Win ratio winRatioPerDay floats.Slice // Profit factor profitFactorPerDay floats.Slice // Trade number dailyTrades floats.Slice accumulatedTrades int previousAccumulatedTrades int } func (r *AccumulatedProfitReport) Initialize() { if r.AccumulatedProfitMAWindow <= 0 { r.AccumulatedProfitMAWindow = 60 } if r.IntervalWindow <= 0 { r.IntervalWindow = 7 } if r.AccumulatedDailyProfitWindow <= 0 { r.AccumulatedDailyProfitWindow = 7 } if r.NumberOfInterval <= 0 { r.NumberOfInterval = 1 } r.accumulatedProfitMA = &indicator.SMA{IntervalWindow: types.IntervalWindow{Interval: types.Interval1d, Window: r.AccumulatedProfitMAWindow}} } func (r *AccumulatedProfitReport) RecordProfit(profit fixedpoint.Value) { r.accumulatedProfit = r.accumulatedProfit.Add(profit) } func (r *AccumulatedProfitReport) RecordTrade(fee fixedpoint.Value) { r.accumulatedFee = r.accumulatedFee.Add(fee) r.accumulatedTrades += 1 } func (r *AccumulatedProfitReport) DailyUpdate(tradeStats *types.TradeStats) { // Daily profit r.dailyProfit.Update(r.accumulatedProfit.Sub(r.previousAccumulatedProfit).Float64()) r.previousAccumulatedProfit = r.accumulatedProfit // Accumulated profit r.accumulatedProfitPerDay.Update(r.accumulatedProfit.Float64()) // Accumulated profit MA r.accumulatedProfitMA.Update(r.accumulatedProfit.Float64()) r.accumulatedProfitMAPerDay.Update(r.accumulatedProfitMA.Last(0)) // Accumulated Fee r.accumulatedFeePerDay.Update(r.accumulatedFee.Float64()) // Win ratio r.winRatioPerDay.Update(tradeStats.WinningRatio.Float64()) // Profit factor r.profitFactorPerDay.Update(tradeStats.ProfitFactor.Float64()) // Daily trades r.dailyTrades.Update(float64(r.accumulatedTrades - r.previousAccumulatedTrades)) r.previousAccumulatedTrades = r.accumulatedTrades } // Output Accumulated profit report to a TSV file func (r *AccumulatedProfitReport) Output(symbol string) { if r.TsvReportPath != "" { tsvwiter, err := tsv.AppendWriterFile(r.TsvReportPath) if err != nil { panic(err) } defer tsvwiter.Close() // Output symbol, total acc. profit, acc. profit 60MA, interval acc. profit, fee, win rate, profit factor _ = tsvwiter.Write([]string{"#", "Symbol", "accumulatedProfit", "accumulatedProfitMA", fmt.Sprintf("%dd profit", r.AccumulatedDailyProfitWindow), "accumulatedFee", "winRatio", "profitFactor", "60D trades"}) for i := 0; i <= r.NumberOfInterval-1; i++ { accumulatedProfit := r.accumulatedProfitPerDay.Index(r.IntervalWindow * i) accumulatedProfitStr := fmt.Sprintf("%f", accumulatedProfit) accumulatedProfitMA := r.accumulatedProfitMAPerDay.Index(r.IntervalWindow * i) accumulatedProfitMAStr := fmt.Sprintf("%f", accumulatedProfitMA) intervalAccumulatedProfit := r.dailyProfit.Tail(r.AccumulatedDailyProfitWindow+r.IntervalWindow*i).Sum() - r.dailyProfit.Tail(r.IntervalWindow*i).Sum() intervalAccumulatedProfitStr := fmt.Sprintf("%f", intervalAccumulatedProfit) accumulatedFee := fmt.Sprintf("%f", r.accumulatedFeePerDay.Index(r.IntervalWindow*i)) winRatio := fmt.Sprintf("%f", r.winRatioPerDay.Index(r.IntervalWindow*i)) profitFactor := fmt.Sprintf("%f", r.profitFactorPerDay.Index(r.IntervalWindow*i)) trades := r.dailyTrades.Tail(60+r.IntervalWindow*i).Sum() - r.dailyTrades.Tail(r.IntervalWindow*i).Sum() tradesStr := fmt.Sprintf("%f", trades) _ = tsvwiter.Write([]string{fmt.Sprintf("%d", i+1), symbol, accumulatedProfitStr, accumulatedProfitMAStr, intervalAccumulatedProfitStr, accumulatedFee, winRatio, profitFactor, tradesStr}) } } } func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) { session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: s.Interval}) if !bbgo.IsBackTesting { session.Subscribe(types.MarketTradeChannel, s.Symbol, types.SubscribeOptions{}) } s.ExitMethods.SetAndSubscribe(session, s) } func (s *Strategy) ID() string { return ID } func (s *Strategy) InstanceID() string { return fmt.Sprintf("%s:%s", ID, s.Symbol) } func (s *Strategy) CalcAssetValue(price fixedpoint.Value) fixedpoint.Value { balances := s.session.GetAccount().Balances() return balances[s.Market.BaseCurrency].Total().Mul(price).Add(balances[s.Market.QuoteCurrency].Total()) } func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, session *bbgo.ExchangeSession) error { var instanceID = s.InstanceID() if s.Position == nil { s.Position = types.NewPositionFromMarket(s.Market) } if s.ProfitStats == nil { s.ProfitStats = types.NewProfitStats(s.Market) } if s.TradeStats == nil { s.TradeStats = types.NewTradeStats(s.Symbol) } // StrategyController s.Status = types.StrategyStatusRunning s.OnSuspend(func() { // Cancel active orders _ = s.orderExecutor.GracefulCancel(ctx) }) s.OnEmergencyStop(func() { // Cancel active orders _ = s.orderExecutor.GracefulCancel(ctx) // Close 100% position //_ = s.ClosePosition(ctx, fixedpoint.One) }) s.session = session // Set fee rate if s.session.MakerFeeRate.Sign() > 0 || s.session.TakerFeeRate.Sign() > 0 { s.Position.SetExchangeFeeRate(s.session.ExchangeName, types.ExchangeFee{ MakerFeeRate: s.session.MakerFeeRate, TakerFeeRate: s.session.TakerFeeRate, }) } s.orderExecutor = bbgo.NewGeneralOrderExecutor(session, s.Symbol, ID, instanceID, s.Position) s.orderExecutor.BindEnvironment(s.Environment) s.orderExecutor.BindProfitStats(s.ProfitStats) s.orderExecutor.BindTradeStats(s.TradeStats) // AccountValueCalculator s.AccountValueCalculator = bbgo.NewAccountValueCalculator(s.session, s.Market.QuoteCurrency) // Accumulated profit report if bbgo.IsBackTesting { if s.AccumulatedProfitReport == nil { s.AccumulatedProfitReport = &AccumulatedProfitReport{} } s.AccumulatedProfitReport.Initialize() s.orderExecutor.TradeCollector().OnProfit(func(trade types.Trade, profit *types.Profit) { if profit == nil { return } s.AccumulatedProfitReport.RecordProfit(profit.Profit) }) // s.orderExecutor.TradeCollector().OnTrade(func(trade types.Trade, profit fixedpoint.Value, netProfit fixedpoint.Value) { // s.AccumulatedProfitReport.RecordTrade(trade.Fee) // }) session.MarketDataStream.OnKLineClosed(types.KLineWith(s.Symbol, types.Interval1d, func(kline types.KLine) { s.AccumulatedProfitReport.DailyUpdate(s.TradeStats) })) } // For drawing profitSlice := floats.Slice{1., 1.} price, _ := session.LastPrice(s.Symbol) initAsset := s.CalcAssetValue(price).Float64() cumProfitSlice := floats.Slice{initAsset, initAsset} s.orderExecutor.TradeCollector().OnTrade(func(trade types.Trade, profit fixedpoint.Value, netProfit fixedpoint.Value) { if bbgo.IsBackTesting { s.AccumulatedProfitReport.RecordTrade(trade.Fee) } // For drawing/charting price := trade.Price.Float64() if s.buyPrice > 0 { profitSlice.Update(price / s.buyPrice) cumProfitSlice.Update(s.CalcAssetValue(trade.Price).Float64()) } else if s.sellPrice > 0 { profitSlice.Update(s.sellPrice / price) cumProfitSlice.Update(s.CalcAssetValue(trade.Price).Float64()) } if s.Position.IsDust(trade.Price) { s.buyPrice = 0 s.sellPrice = 0 s.highestPrice = 0 s.lowestPrice = 0 } else if s.Position.IsLong() { s.buyPrice = price s.sellPrice = 0 s.highestPrice = s.buyPrice s.lowestPrice = 0 } else { s.sellPrice = price s.buyPrice = 0 s.highestPrice = 0 s.lowestPrice = s.sellPrice } }) s.InitDrawCommands(&profitSlice, &cumProfitSlice) s.orderExecutor.TradeCollector().OnPositionUpdate(func(position *types.Position) { bbgo.Sync(ctx, s) }) s.orderExecutor.Bind() for _, method := range s.ExitMethods { method.Bind(session, s.orderExecutor) } kLineStore, _ := s.session.MarketDataStore(s.Symbol) s.shark = &SHARK{IntervalWindow: types.IntervalWindow{Window: s.Window, Interval: s.Interval}} s.shark.BindK(s.session.MarketDataStream, s.Symbol, s.shark.Interval) if klines, ok := kLineStore.KLinesOfInterval(s.shark.Interval); ok { s.shark.LoadK((*klines)[0:]) } states := types.NewQueue(s.Window) states.Update(0) s.session.MarketDataStream.OnKLineClosed(types.KLineWith(s.Symbol, s.Interval, func(kline types.KLine) { log.Infof("shark score: %f, current price: %f", s.shark.Last(0), kline.Close.Float64()) nextState := hmm(s.shark.Array(s.Window), states.Array(s.Window), s.Window) states.Update(nextState) log.Infof("Denoised signal via HMM: %f", states.Last(0)) if states.Length() < s.Window { return } direction := 0. if s.Position.IsLong() { direction = 1. } else if s.Position.IsShort() { direction = -1. } if s.Position.IsOpened(kline.Close) && states.Mean(5) == 0 { s.orderExecutor.ClosePosition(ctx, fixedpoint.One) } if states.Mean(5) == 1 && direction != 1 { _, _ = s.orderExecutor.SubmitOrders(ctx, types.SubmitOrder{ Symbol: s.Symbol, Side: types.SideTypeBuy, Quantity: s.Quantity, Type: types.OrderTypeMarket, Tag: "sharkLong", }) } else if states.Mean(5) == -1 && direction != -1 { _, _ = s.orderExecutor.SubmitOrders(ctx, types.SubmitOrder{ Symbol: s.Symbol, Side: types.SideTypeSell, Quantity: s.Quantity, Type: types.OrderTypeMarket, Tag: "sharkShort", }) } })) bbgo.OnShutdown(ctx, func(ctx context.Context, wg *sync.WaitGroup) { defer wg.Done() // Output accumulated profit report if bbgo.IsBackTesting { defer s.AccumulatedProfitReport.Output(s.Symbol) if s.DrawGraph { if err := s.Draw(&profitSlice, &cumProfitSlice); err != nil { log.WithError(err).Errorf("cannot draw graph") } } } _, _ = fmt.Fprintln(os.Stderr, s.TradeStats.String()) _ = s.orderExecutor.GracefulCancel(ctx) }) return nil } // TODO: dirichlet distribution is a too naive solution func observeDistribution(y_t, x_t float64) float64 { if x_t == 0. && y_t == 0 { // observed zero value from indicator when in neutral state return 1. } else if x_t > 0. && y_t > 0. { // observed positive value from indicator when in long state return 1. } else if x_t < 0. && y_t < 0. { // observed negative value from indicator when in short state return 1. } else { return 0. } } func transitProbability(x_t0, x_t1 int) float64 { // stick to the same sate if x_t0 == x_t1 { return 0.99 } // transit to next new state return 1 - 0.99 } // HMM main function, ref: https://tr8dr.github.io/HMMFiltering/ /* # initialize time step 0 using state priors and observation dist p(y | x = s) for si in states: alpha[t = 0, state = si] = pi[si] * p(y[0] | x = si) # determine alpha for t = 1 .. n for t in 1 .. n: for sj in states: alpha[t,sj] = max([alpha[t-1,si] * M[si,sj] for si in states]) * p(y[t] | x = sj) # determine current state at time t return argmax(alpha[t,si] over si) */ func hmm(y_t []float64, x_t []float64, l int) float64 { al := make([]float64, l) an := make([]float64, l) as := make([]float64, l) long := 0. neut := 0. short := 0. // n is the incremental time steps for n := 2; n <= len(x_t); n++ { for j := -1; j <= 1; j++ { sil := make([]float64, 3) sin := make([]float64, 3) sis := make([]float64, 3) for i := -1; i <= 1; i++ { sil = append(sil, x_t[n-1-1]*transitProbability(i, j)) sin = append(sin, x_t[n-1-1]*transitProbability(i, j)) sis = append(sis, x_t[n-1-1]*transitProbability(i, j)) } if j > 0 { _, longArr := floats.MinMax(sil, 3) long = longArr[0] * observeDistribution(y_t[n-1], float64(j)) al = append(al, long) } else if j == 0 { _, neutArr := floats.MinMax(sin, 3) neut = neutArr[0] * observeDistribution(y_t[n-1], float64(j)) an = append(an, neut) } else if j < 0 { _, shortArr := floats.MinMax(sis, 3) short = shortArr[0] * observeDistribution(y_t[n-1], float64(j)) as = append(as, short) } } } _, maximum := floats.MinMax([]float64{long, neut, short}, 3) if maximum[0] == long { return 1 } else if maximum[0] == short { return -1 } return 0 }