374 lines
10 KiB
Go
374 lines
10 KiB
Go
package arb
|
||
|
||
import (
|
||
"fmt"
|
||
"github.com/banbox/banbot/config"
|
||
"github.com/banbox/banbot/core"
|
||
"github.com/banbox/banbot/strat"
|
||
"github.com/banbox/banexg/log"
|
||
"github.com/banbox/banta"
|
||
"go.uber.org/zap"
|
||
"math"
|
||
"sort"
|
||
"strconv"
|
||
"sync"
|
||
)
|
||
|
||
/*
|
||
统计套利：对两个交易对的价格比率下注
|
||
价格比率较稳定的：
|
||
快速穿越大趋势跟随开单
|
||
超出上下轨，回弹交叉开单
|
||
价格比率明显单边趋势：
|
||
在标准差30%以内交叉，跟随大趋势方向开单
|
||
|
||
强劲单边跟随：
|
||
平时价格比率小幅上下波动，但偶尔有强劲单边价格剧烈运动：GLM/USDT:USDT, XEM/USDT:USDT 20241001
|
||
|
||
瞬时极端波动：
|
||
某段时间，价格比率快速上下极端波动，GLM/USDT:USDT, XEM/USDT:USDT 20241025
|
||
|
||
*/
|
||
|
||
func MeanRegressStat(p *config.RunPolicyConfig) *strat.TradeStrat {
|
||
var sta *StatTwo
|
||
var jobs = make(map[string]*strat.StratJob)
|
||
return &strat.TradeStrat{
|
||
WarmupNum: 1000,
|
||
BatchInOut: true,
|
||
OnStartUp: func(s *strat.StratJob) {
|
||
jobs[s.Symbol.Symbol] = s
|
||
},
|
||
OnBar: func(s *strat.StratJob) {
|
||
if sta == nil {
|
||
sta = newStatTwo(jobs, nil)
|
||
}
|
||
e := s.Env
|
||
if e.TimeStart > sta.barMS {
|
||
sta.barMS = e.TimeStart
|
||
}
|
||
sta.OnPrice(s.Symbol.Symbol, e.Close.Get(0))
|
||
},
|
||
OnBatchJobs: func(jobs []*strat.StratJob) {
|
||
sta.onStampDone()
|
||
},
|
||
}
|
||
}
|
||
|
||
const (
|
||
slopeDist = 150
|
||
maSlopeDist = 500
|
||
)
|
||
|
||
type StatTwo struct {
|
||
barMS int64 // 收到最新bar的时间戳
|
||
prices map[string]float64 // 当前时间戳下，所有品种的价格
|
||
jobs map[string]*strat.StratJob // 所有的任务
|
||
cacheSize int // Tuple中保留历史数据的最小长度
|
||
lock sync.Mutex
|
||
Tuples map[string]map[string]*Tuple
|
||
}
|
||
|
||
type Tuple struct {
|
||
A *strat.StratJob
|
||
B *strat.StratJob
|
||
env *banta.BarEnv
|
||
raw *banta.Series
|
||
maFast *banta.Series
|
||
maSlow *banta.Series
|
||
maBig *banta.Series
|
||
rawSbBig *banta.Series // sma(raw-maBig, 500)
|
||
faXSl int // fast slow交叉：1快线上穿慢线；-1快线下穿慢线
|
||
rawXBig int // raw big交叉
|
||
maSlope float64 // ma斜率（和500个之前）
|
||
slope float64 // 价格比率斜率（和150个之前）
|
||
sort float64 // 排序分数，降序
|
||
stdDev float64 // 标准差
|
||
allowType int // 允许开单类型: 1突破布林带回弹；2快速穿越大趋势线；3中心交叉跟随趋势
|
||
allowDirt int // 允许开单方向: 1做多A做空B； -1做空A做多B
|
||
openType int // 开单类型：从allowType赋值
|
||
openRate float64 // 开单的价格比率，0表示未开单
|
||
openAt int64 // 开单的时间戳
|
||
}
|
||
|
||
func newStatTwo(jobs map[string]*strat.StratJob, tuples [][]string) *StatTwo {
|
||
tupleMap := make(map[string]map[string]*Tuple)
|
||
res := &StatTwo{
|
||
cacheSize: 2000,
|
||
jobs: jobs,
|
||
Tuples: tupleMap,
|
||
}
|
||
if len(tuples) == 0 {
|
||
codes := make([]string, 0, len(jobs))
|
||
for key := range jobs {
|
||
codes = append(codes, key)
|
||
}
|
||
for i := 0; i < len(codes); i++ {
|
||
for j := i + 1; j < len(codes); j++ {
|
||
tuples = append(tuples, []string{codes[i], codes[j]})
|
||
}
|
||
}
|
||
}
|
||
if len(tuples) > 7000 {
|
||
log.Warn("too many tuples may cause OOM!", zap.Int("num", len(tuples)))
|
||
}
|
||
for _, va := range tuples {
|
||
a, b := va[0], va[1]
|
||
if a > b {
|
||
a, b = b, a
|
||
}
|
||
subMap, ok := tupleMap[a]
|
||
if !ok {
|
||
subMap = make(map[string]*Tuple)
|
||
tupleMap[a] = subMap
|
||
}
|
||
tup, ok := subMap[b]
|
||
if !ok {
|
||
baseA, _, _, _ := core.SplitSymbol(a)
|
||
baseB, _, _, _ := core.SplitSymbol(b)
|
||
symbol := fmt.Sprintf("%s/%s", baseA, baseB)
|
||
env, _ := banta.NewBarEnv(core.ExgName, core.Market, symbol, "1m")
|
||
env.MaxCache = res.cacheSize
|
||
raw := env.NewSeries(nil)
|
||
A, _ := jobs[a]
|
||
B, _ := jobs[b]
|
||
tup = &Tuple{
|
||
A: A,
|
||
B: B,
|
||
raw: raw,
|
||
env: env,
|
||
}
|
||
subMap[b] = tup
|
||
}
|
||
}
|
||
return res
|
||
}
|
||
|
||
func (s *StatTwo) OnPrice(symbol string, price float64) {
|
||
s.lock.Lock()
|
||
s.prices[symbol] = price
|
||
s.lock.Unlock()
|
||
}
|
||
|
||
func (s *StatTwo) popTuples() []*Tuple {
|
||
s.lock.Lock()
|
||
tasks := make(map[string]map[string]bool)
|
||
for codeA := range s.prices {
|
||
subMap := make(map[string]bool)
|
||
for codeB := range s.prices {
|
||
if codeA >= codeB {
|
||
continue
|
||
}
|
||
subMap[codeB] = true
|
||
}
|
||
tasks[codeA] = subMap
|
||
}
|
||
var res = make([]*Tuple, 0, len(s.prices)*2)
|
||
for codeA, subMap := range tasks {
|
||
for codeB := range subMap {
|
||
ratio := s.prices[codeA] / s.prices[codeB]
|
||
t := s.tuple(codeA, codeB, ratio)
|
||
if t == nil {
|
||
continue
|
||
}
|
||
res = append(res, t)
|
||
}
|
||
}
|
||
s.prices = make(map[string]float64)
|
||
s.lock.Unlock()
|
||
return res
|
||
}
|
||
|
||
func (s *StatTwo) onStampDone() {
|
||
tuples := s.popTuples()
|
||
// 筛选两种开单类型
|
||
size := len(tuples) / 3
|
||
list1 := make([]*Tuple, 0, size)
|
||
list2 := make([]*Tuple, 0, size)
|
||
list3 := make([]*Tuple, 0, size)
|
||
for _, t := range tuples {
|
||
t.calcInOuts()
|
||
if t.allowType == 1 {
|
||
list1 = append(list1, t)
|
||
} else if t.allowType == 2 {
|
||
list2 = append(list2, t)
|
||
} else if t.allowType == 3 {
|
||
list3 = append(list3, t)
|
||
}
|
||
}
|
||
var list = list1
|
||
if len(list) == 0 {
|
||
if len(list2) == 0 {
|
||
if len(list3) == 0 {
|
||
return
|
||
}
|
||
list = list3
|
||
} else {
|
||
list = list2
|
||
}
|
||
}
|
||
// 按分数降序，取前三个中没有开单的
|
||
var it *Tuple
|
||
sort.Slice(list, func(i, j int) bool {
|
||
return list[i].sort > list[j].sort
|
||
})
|
||
gap := len(list) / 2
|
||
list = append(list[gap:], list[:gap]...)
|
||
for i, v := range list {
|
||
if v.openType == 0 && v.A.OrderNum == 0 && v.B.OrderNum == 0 {
|
||
it = v
|
||
break
|
||
}
|
||
if i == 2 {
|
||
break
|
||
}
|
||
}
|
||
if it == nil {
|
||
return
|
||
}
|
||
// 下单
|
||
shortA := it.allowDirt < 0
|
||
errA := it.A.OpenOrder(&strat.EnterReq{
|
||
Tag: strconv.Itoa(it.allowType),
|
||
Short: shortA,
|
||
})
|
||
if errA == nil {
|
||
errB := it.B.OpenOrder(&strat.EnterReq{
|
||
Tag: strconv.Itoa(it.allowType),
|
||
Short: !shortA,
|
||
})
|
||
if errB != nil {
|
||
it.A.Entrys = nil
|
||
} else {
|
||
it.openType = it.allowType
|
||
it.openRate = it.raw.Get(0)
|
||
it.openAt = it.env.TimeStop
|
||
}
|
||
}
|
||
//dateStr := btime.ToDateStr(s.barMS, core.DefaultDateFmt)
|
||
//fmt.Printf("%s %s - %s type=%v sort=%.3f slope=%.3f bigSlope=%.3f\n", dateStr, it.A.Symbol.Symbol,
|
||
// it.B.Symbol.Symbol, it.allowType, it.sort, it.slope, it.maSlope)
|
||
}
|
||
|
||
func (s *StatTwo) tuple(a, b string, ratio float64) *Tuple {
|
||
if a > b {
|
||
a, b = b, a
|
||
ratio = 1 / ratio
|
||
}
|
||
subMap, ok := s.Tuples[a]
|
||
if !ok {
|
||
return nil
|
||
}
|
||
res, ok := subMap[b]
|
||
if ok {
|
||
res.update(s.barMS, ratio)
|
||
return res
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (t *Tuple) update(stamp int64, ratio float64) {
|
||
t.env.TimeStart = stamp
|
||
t.env.TimeStop = stamp + t.env.TFMSecs
|
||
t.env.BarNum += 1
|
||
t.raw.Append(ratio)
|
||
t.maFast = banta.EMA(t.raw, 24)
|
||
t.maSlow = banta.EMA(t.raw, 66)
|
||
t.maBig = banta.EMA(t.raw, 1333)
|
||
t.rawSbBig = banta.SMA(t.raw.Sub(t.maBig), 500)
|
||
t.faXSl = t.maFast.Cross(t.maSlow)
|
||
t.rawXBig = t.raw.Cross(t.maBig)
|
||
sdev := banta.StdDev(t.raw, 1333)
|
||
t.stdDev = banta.SMA(sdev, 800).Get(0) * 1.7
|
||
if t.raw.Len() > slopeDist {
|
||
// 计算近期斜率，优先用基于历史150偏移计算
|
||
// 当基于均线历史150偏移的绝对变动区间更小时，基于均线计算
|
||
old := t.raw.Get(slopeDist)
|
||
oldMa := t.maBig.Get(slopeDist)
|
||
slope := ratio/old - 1
|
||
slopeMa := ratio/oldMa - 1
|
||
if math.Abs(slopeMa) < math.Abs(slope) {
|
||
slope = slopeMa
|
||
}
|
||
t.slope = slope
|
||
}
|
||
if t.maBig.Len() > maSlopeDist {
|
||
t.allowType = 0
|
||
t.sort = 0
|
||
t.maSlope = t.maBig.Get(0)/t.maBig.Get(maSlopeDist) - 1
|
||
}
|
||
}
|
||
|
||
func (t *Tuple) calcInOuts() {
|
||
if t.maBig.Len() < maSlopeDist {
|
||
return
|
||
}
|
||
//fastVal, fastVal30 := t.maFast.Get(0), t.maFast.Get(30)
|
||
//if t.rawXBig == -1 || t.rawXBig == 1 {
|
||
// // 快速穿越大趋势，开单
|
||
// fastSlope := math.Abs(fastVal/fastVal30 - 1)
|
||
// distRate := t.rawSbBig.Get(0) / stdDev
|
||
// if fastSlope > 0.01 && distRate*float64(t.rawXBig) < 0 && math.Abs(distRate) > 0.7 {
|
||
// // 要求穿过大趋势线时，具备较高的斜率，且过去500个平均与大周期的差超过70%标准差
|
||
// t.allowType = 2
|
||
// t.allowDirt = t.rawXBig
|
||
// t.sort = math.Abs(distRate)
|
||
// }
|
||
//}
|
||
maSlopeAbs := math.Abs(t.maSlope)
|
||
//diff := math.Abs(t.slope - t.maSlope)
|
||
//if maSlopeAbs < min(math.Abs(t.slope), 0.02) && diff > 0.015 {
|
||
// // 突破布林带上下轨，回弹时开单
|
||
// bigVal := t.maBig.Get(0)
|
||
// slowPrev := t.maSlow.Get(60)
|
||
// // 按maSlope进行惩罚，极端0.02时削减64%
|
||
// sortV := diff * math.Pow(1-maSlopeAbs*20, 2)
|
||
// if t.faXSl == -1 && max(fastVal, fastVal30) > bigVal+stdDev {
|
||
// // 均线上方，突破上界，快线下穿慢线
|
||
// if fastVal > slowPrev {
|
||
// t.allowType = 1
|
||
// t.allowDirt = t.faXSl
|
||
// t.sort = sortV
|
||
// }
|
||
// } else if t.faXSl == 1 && min(fastVal, fastVal30) < bigVal-stdDev {
|
||
// // 均线下方，突破下界，快线上穿慢线
|
||
// if fastVal < slowPrev {
|
||
// t.allowType = 1
|
||
// t.allowDirt = t.faXSl
|
||
// t.sort = sortV
|
||
// }
|
||
// }
|
||
//}
|
||
nearToCenter := math.Abs(t.raw.Get(0)-t.maBig.Get(0)) < t.stdDev*0.3
|
||
if maSlopeAbs > 0.02 && nearToCenter && (t.faXSl == 1 || t.faXSl == -1) && t.maSlope*float64(t.faXSl) > 0 {
|
||
// 明显大趋势，在中心附近交叉，按大趋势方向开单
|
||
t.allowType = 3
|
||
t.allowDirt = t.faXSl
|
||
t.sort = maSlopeAbs
|
||
}
|
||
closeOd := false
|
||
if t.openType == 2 {
|
||
// 根据穿越大趋势开单：回穿大趋势平仓、快线慢线交叉平仓
|
||
if t.faXSl*t.allowDirt == -1 || t.rawXBig*t.allowDirt == -1 {
|
||
closeOd = true
|
||
}
|
||
} else if t.openType == 1 && t.faXSl*t.allowDirt == -1 {
|
||
// 根据突破上下轨回弹开单，(盈利达到标准差30% || 持有时间超过700个bar)：快线慢线交叉平仓
|
||
pft30 := (t.raw.Get(0)-t.openRate)*float64(t.allowDirt) > t.stdDev*0.3
|
||
//costBarNum := int((t.env.TimeStop - t.openAt) / t.env.TFMSecs)
|
||
if pft30 { //|| costBarNum > 1000 {
|
||
closeOd = true
|
||
}
|
||
} else if t.openType == 3 && t.faXSl*t.allowDirt == -1 {
|
||
// 中心附近交叉，跟随趋势的，再次交叉时平仓
|
||
closeOd = true
|
||
}
|
||
if closeOd {
|
||
t.A.CloseOrders(&strat.ExitReq{Tag: "exit"})
|
||
t.B.CloseOrders(&strat.ExitReq{Tag: "exit"})
|
||
t.openType = 0
|
||
t.openRate = 0
|
||
t.openAt = 0
|
||
}
|
||
}
|