PorterandCursor
2c1232555e
将核心结构、指标与分析拆到 chan/{core,indicators,analysis,pipeline}；
根目录保留兼容 shim；strategies 改为从 chan 包导入；买卖点经 bsp_macd 与 MACD 接合。
Co-authored-by: Cursor <cursoragent@cursor.com>
91 lines
3.6 KiB
Python
91 lines
3.6 KiB
Python
import copy
|
|
from typing import Dict, Optional
|
|
|
|
from .ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR
|
|
from . import ChanKLU
|
|
from .ChanBI import ChanBI
|
|
|
|
class ChanSBI():
|
|
def __init__(self, start_bi: ChanBI, index, dir=Chan_BI_DIR.UP):
|
|
self.start_bi = start_bi
|
|
self.end_bi = None
|
|
self.index = index
|
|
self.dir = dir
|
|
self.high = start_bi.high
|
|
self.low = start_bi.low
|
|
self.pre = None
|
|
self.next = None
|
|
self.fx = Chan_FX_TYPE.UNKNOWN
|
|
self.bi_list = []
|
|
self.bi_list.append(start_bi)
|
|
self.has_fx_gap = False
|
|
def set_fx(self, fx):
|
|
self.fx = fx
|
|
def set_end_bi(self, bi):
|
|
self.end_bi = bi
|
|
def set_pre(self, sbi):
|
|
self.pre = sbi
|
|
def set_next(self, sbi):
|
|
self.next = sbi
|
|
def add_bi(self, bi):
|
|
self.bi_list.append(bi)
|
|
def check_fx(self):
|
|
if self.pre and self.next:
|
|
#print(self.pre.start_bi.start_time, self.start_bi.start_time, self.end_bi.end_time, self.next.start_bi.start_time, self.pre.high, self.high, self.next.high, self.pre.low, self.low, self.next.low, self.dir)
|
|
if self.high > self.pre.high and self.high > self.next.high:
|
|
self.fx = Chan_FX_TYPE.TOP
|
|
#print(self.start_bi.start_time, self.pre.start_bi.start_time, self.next.start_bi.start_time, self.fx)
|
|
if self.low > self.pre.high:
|
|
self.has_fx_gap = True
|
|
#print(self.start_bi.start_time, self.end_bi.end_time, self.pre.start_bi.start_time, self.next.start_bi.start_time, self.dir, self.has_fx_gap, self.fx)
|
|
return Chan_FX_TYPE.TOP
|
|
else:
|
|
if self.low < self.pre.low and self.low < self.next.low:
|
|
self.fx = Chan_FX_TYPE.BOTTOM
|
|
#print(self.start_bi.start_time, self.pre.start_bi.start_time, self.next.start_bi.start_time, self.fx)
|
|
if self.high < self.pre.low:
|
|
self.has_fx_gap = True
|
|
#print(self.start_bi.start_time, self.end_bi.end_time, self.pre.start_bi.start_time, self.next.start_bi.start_time, self.dir, self.has_fx_gap, self.fx)
|
|
return Chan_FX_TYPE.BOTTOM
|
|
return Chan_FX_TYPE.UNKNOWN
|
|
def check_seg_bi_broken(self):
|
|
broken = False
|
|
if self.fx == Chan_FX_TYPE.TOP:
|
|
if self.next.low < self.pre.high:
|
|
broken = True
|
|
elif self.fx == Chan_FX_TYPE.BOTTOM:
|
|
if self.next.high > self.pre.low:
|
|
broken = True
|
|
return broken
|
|
def check_bi_included(self, bi):
|
|
included = False
|
|
if self.high > bi.high:
|
|
# high大于，low小于，左包含
|
|
if self.low < bi.low:
|
|
included = True
|
|
# high大于，low大于，不包含
|
|
else:
|
|
# if self.low > bi.low
|
|
# high相等，右包含
|
|
included = False
|
|
else:
|
|
included = False
|
|
# high小于，low大于，右包含
|
|
#if self.low > bi.low:
|
|
#included = True
|
|
if included:
|
|
if self.pre:
|
|
if self.high > self.pre.high and self.low < self.pre.low:
|
|
included = True
|
|
if included:
|
|
self.add_bi(bi)
|
|
# gn>gn-1
|
|
if self.dir == Chan_BI_DIR.DOWN:
|
|
# UP -> max(dn)
|
|
self.low = bi.low
|
|
else:
|
|
# DOWN -> min(gn)
|
|
self.high = bi.high
|
|
#self.print(bi, "Z")
|
|
#print(self.start_bi.start_time, bi.start_time, included)
|
|
return included |