def intervalCompute(intvA, intvB):
# 计算A、B区间是否重叠,如有,则返回重叠部分,否则返回方向关系(b > a则为up反之down),和None
if intvA.overlaps(intvB):
intvRet = Interval(max((intvA.lower_bound, intvB.lower_bound)), min((intvA.upper_bound, intvB.upper_bound)))
return "overlap", intvRet
elif intvA.upper_bound < intvB.lower_bound:
return "up", None
else:
return "down", None
def isIncluding(intvA, intvB):
# 计算AB是否为包含关系,只要一方包含另一方,则返回True, 否则False
isOverlap, intvOvlp = intervalCompute(intvA, intvB)
if isOverlap == "overlap" and intvOvlp == intvA:
# 重复段为前者,即后包前
return True, 0
if isOverlap == "overlap" and intvOvlp == intvB:
# 重复段为后者,即前包后
return True, 1
else:
# 无包含关系
return False
, None
def includingProcess(intvA, intvB, direction="up"):
# 计算相互有包含关系的A、B合并后的新区间,拟缠论规则
if direction == "up":
return Interval(max((intvA.lower_bound, intvB.lower_bound)), max((intvA.upper_bound, intvB.upper_bound)))
elif direction == "down":
return Interval(min((intvA.lower_bound, intvB.lower_bound)), min((intvA.upper_bound, intvB.upper_bound)))
def _reviseInclude(dfKlinesCopy, intvRet, close, iloc):
# 辅助函数,用于最后按包含关系修改dfKlinesCopy
dfKlinesCopy["HIGH"][iloc] = intvRet.upper_bound
dfKlinesCopy["LOW"][iloc] = intvRet.lower_bound
dfKlinesCopy["CLOSE"][iloc] = close
return dfKlinesCopy