gavindiaz/chanlun.rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

首次提交

Browse Source
This commit is contained in:
YuWuKunCheng
2026-06-09 20:09:55 +08:00
parent 15a1d43b1d
commit 8405d478bc
11 changed files with 6663 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
LICENSES/Apache 2.0 - czsc
+13
View File
@@ -0,0 +1,13 @@
Copyright [2025] [zengbin93]
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
LICENSES/MIT - parse
+19
View File
@@ -0,0 +1,19 @@
Copyright (c) 2012-2019 Richard Jones <richard@python.org>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
LICENSES/MIT - termcolor
+19
View File
@@ -0,0 +1,19 @@
Copyright (c) 2008-2011 Volvox Development Team
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
NOTICE
+32
View File
@@ -0,0 +1,32 @@
chanlun — 缠论技术分析库
===========================
Copyright (c) 2026 YuYuKunKun
This product includes software developed by third-party open source projects:
----------------------------------------------------------------------
1. czsc
Repository: <https://github.com/waditu/czsc>
License: Apache License 2.0
Copyright (c) 2025 zengbin93
Used in: chanlun-py/chanlun/chan_external.py（部分代码片段）
----------------------------------------------------------------------
2. parse
Repository: <https://github.com/r1chardj0n3s/parse>
License: MIT License
Copyright (c) 2012-2019 Richard Jones <richard@python.org>
Used in: chanlun-py/chanlun/parse.py
----------------------------------------------------------------------
3. termcolor
Repository: <https://github.com/termcolor/termcolor>
License: MIT License
Copyright (c) 2008-2011 Volvox Development Team
Used in: chanlun-py/chanlun/termcolor.py
----------------------------------------------------------------------
Full licenses are available in the LICENSES/ directory.
chanlun-py/chanlun/chan_external.py
+2286
View File
File diff suppressed because it is too large Load Diff
chanlun-py/chanlun/indicators.py
+1594
View File
File diff suppressed because it is too large Load Diff
chanlun-py/chanlun/parse.py
+1087
View File
File diff suppressed because it is too large Load Diff
chanlun-py/chanlun/parse.pyi
+113
View File
@@ -0,0 +1,113 @@
# Copyright (c) 2012-2019 Richard Jones <richard@python.org>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import re
from typing import Any, Callable, Generic, Literal, Protocol, TypeVar, overload
__all__ = ["parse", "search", "findall", "with_pattern"]
_T = TypeVar("_T")
_T_co = TypeVar("_T_co", covariant=True)
class _TypeConverter(Protocol[_T_co]):
def __call__(self, string: str) -> _T_co: ...
_TTypeConverter = TypeVar("_TTypeConverter", bound="_TypeConverter[Any]")
def with_pattern(pattern: str, regex_group_count=None) -> Callable[[_TTypeConverter], _TTypeConverter]: ...
class Result:
fixed: tuple[Any, ...]
named: dict[str, Any]
spans: dict[int | str, tuple[int, int]]
def __init__(self, fixed: tuple[Any, ...], named: dict[str, Any], spans: dict[int | str, tuple[int, int]]) -> None: ...
def __getitem__(self, item) -> Any: ...
def __contains__(self, name) -> bool: ...
class Match:
parser: "Parser"
match: re.Match # type: ignore[type-arg]
def __init__(self, parser: "Parser", match: re.Match) -> None: ... # type: ignore[type-arg]
def evaluate_result(self) -> Result: ...
class ResultIterator(Generic[_T]):
parser: "Parser"
string: str
pos: int
endpos: int
evaluate_result: bool
def __next__(self) -> _T: ...
next = __next__
def __init__(self, parser: "Parser", string: str, pos: int, endpos: int | None, evaluate_result: bool = True) -> None: ...
def __iter__(self) -> "ResultIterator[_T]": ...
class TooManyFields(ValueError): ...
class RepeatedNameError(ValueError): ...
class Parser:
def __init__(self, format: str, extra_types: dict[str, _TypeConverter[Any]] | None = None, case_sensitive: bool = False) -> None: ...
@property
def named_fields(self) -> list[str]: ...
@property
def fixed_fields(self) -> list[int]: ...
@property
def format(self) -> str: ...
@overload
def parse(self, string: str, evaluate_result: Literal[True] = True) -> Result | None: ...
@overload
def parse(self, string: str, *, evaluate_result: Literal[False]) -> Match | None: ...
@overload
def parse(self, string: str, evaluate_result: Literal[False]) -> Match | None: ...
@overload
def search(self, string: str, pos: int = 0, endpos: int | None = None, evaluate_result: Literal[True] = True) -> Result | None: ...
@overload
def search(self, string: str, pos: int = 0, endpos: int | None = None, *, evaluate_result: Literal[False]) -> Match | None: ...
@overload
def search(self, string: str, pos: int, endpos: int | None, evaluate_result: Literal[False]) -> Match | None: ...
@overload
def findall(self, string: str, pos: int = 0, endpos=None, extra_types: dict[str, _TypeConverter[Any]] | None = None, evaluate_result: Literal[True] = True) -> ResultIterator[Result]: ...
@overload
def findall(self, string: str, pos: int = 0, endpos=None, extra_types: dict[str, _TypeConverter[Any]] | None = None, *, evaluate_result: Literal[False]) -> ResultIterator[Match]: ...
@overload
def findall(self, string: str, pos: int, endpos: int | None, extra_types, evaluate_result: Literal[False]) -> ResultIterator[Match]: ...
def evaluate_result(self, m: re.Match) -> Result: ... # type: ignore[type-arg]
@overload
def parse(format: str, string: str, extra_types: dict[str, _TypeConverter[Any]] | None = None, evaluate_result: Literal[True] = True, case_sensitive: bool = ...) -> Result | None: ...
@overload
def parse(format: str, string: str, extra_types: dict[str, _TypeConverter[Any]] | None = None, *, evaluate_result: Literal[False], case_sensitive: bool = ...) -> Match | None: ...
@overload
def parse(format: str, string: str, extra_types, evaluate_result: Literal[False], case_sensitive: bool = ...) -> Match | None: ...
@overload
def search(format: str, string: str, pos: int = 0, endpos: int | None = None, extra_types: dict[str, _TypeConverter[Any]] | None = None, evaluate_result: Literal[True] = True, case_sensitive: bool = False) -> Result | None: ...
@overload
def search(format: str, string: str, pos: int = 0, endpos: int | None = None, extra_types: dict[str, _TypeConverter[Any]] | None = None, *, evaluate_result: Literal[False], case_sensitive: bool = False) -> Match | None: ...
@overload
def search(format: str, string: str, pos: int, endpos: int | None, extra_types, evaluate_result: Literal[False], case_sensitive: bool = False) -> Match | None: ...
@overload
def findall(format: str, string: str, pos: int = 0, endpos=None, extra_types: dict[str, _TypeConverter[Any]] | None = None, evaluate_result: Literal[True] = True, case_sensitive: bool = False) -> ResultIterator[Result]: ...
@overload
def findall(format: str, string: str, pos: int = 0, endpos=None, extra_types: dict[str, _TypeConverter[Any]] | None = None, *, evaluate_result: Literal[False], case_sensitive: bool = False) -> ResultIterator[Match]: ...
@overload
def findall(format, string, pos, endpos, extra_types, evaluate_result: Literal[False], case_sensitive: bool = False) -> ResultIterator[Match]: ...
def compile(format: str, extra_types: dict[str, _TypeConverter[Any]] | None = None, case_sensitive: bool = False) -> Parser: ...
chanlun-py/chanlun/signals/__init__.py
+506
View File
@@ -0,0 +1,506 @@
"""缠论技术分析库 — 信号函数模块
每个信号函数接收 观察者 对象 + 关键字参数，返回 OrderedDict。
信号 key 格式：k1_k2_k3，value 格式：v1_v2_v3_score。
数据访问路径：
- K线指标：k线.指标.macd.DIF / k线.指标.rsi.RSI / k线.指标.kdj.K / k线.指标.均线["SMA_5"]
- 笔序列：观察员.笔序列（List[虚线]）
- 分型序列：观察员.分型序列（List[分型]）
"""
from collections import OrderedDict
from chanlun.chan import 观察者, 分型结构, 虚线, 相对方向
from chanlun.chan_external import create_single_signal
# 信号函数模板
def 模板_V日期(观察员: 观察者, **kwargs) -> OrderedDict:
"""##信号名称介绍##
触发条件：## 触发条件 ##
参数模板：## 具体模板 如: "{freq}_D{di}#{ma_type}#{timeperiod}MO{max_overlap}_BS辅助V230313" ##
**信号逻辑：**
## 详细信号逻辑 ##
**信号列表：**
## 具体信号 如下:
- Signal('15分钟_D1#SMA#5MO5_BS辅助V230313_看空_向下_任意_0')
- Signal('15分钟_D1#SMA#5MO5_BS辅助V230313_看多_向下_任意_0')
- Signal('15分钟_D1#SMA#5MO5_BS辅助V230313_看多_向上_任意_0')
- Signal('15分钟_D1#SMA#5MO5_BS辅助V230313_看空_向上_任意_0')
##
:param 观察员: 观察者对象
:param kwargs: 其他参数
- ## 具体参数介绍 ##
:return: 信号识别结果
"""
## 具体代码过程 ##
return ## create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2) ##
# ==============================================================================
# tas — 技术指标信号
# ==============================================================================
def tas_ma_base_V230313(c, **kwargs) -> OrderedDict:
"""单均线多空和方向辅助开平仓信号
参数模板："{freq}_D{di}#{ma_type}#{timeperiod}MO{max_overlap}_BS辅助V230313"
**信号逻辑：**
1. close > ma，多头（看多）；反之，空头（看空）
2. ma[-1] > ma[-2]，向上；反之，向下
3. 加入 max_overlap 参数控制相同信号最大重叠次数
**信号列表：**
- Signal('15分钟_D1#SMA#5MO5_BS辅助V230313_看空_向下_任意_0')
- Signal('15分钟_D1#SMA#5MO5_BS辅助V230313_看多_向下_任意_0')
- Signal('15分钟_D1#SMA#5MO5_BS辅助V230313_看多_向上_任意_0')
- Signal('15分钟_D1#SMA#5MO5_BS辅助V230313_看空_向上_任意_0')
:param c: 观察者对象
:param kwargs: 其他参数
- ma_type: 均线类型（SMA/EMA）
- timeperiod: 均线计算周期
- di: 信号计算截止倒数第i根K线
- max_overlap: 相同信号最大重叠次数
:return: 信号识别结果
"""
ma_type = kwargs.get("ma_type", "SMA").upper()
timeperiod = int(kwargs.get("timeperiod", 5))
di = int(kwargs.get("di", 1))
max_overlap = int(kwargs.get("max_overlap", 5))
freq = kwargs.get("freq", "15分钟")
k1, k2, k3 = f"{freq}_D{di}#{ma_type}#{timeperiod}MO{max_overlap}_BS辅助V230313".split("_", 2)
普K序列 = c.普通K线序列
if len(普K序列) < di + 1:
return create_single_signal(k1=k1, k2=k2, k3=k3)
当前K线 = 普K序列[-di]
if 当前K线.指标 is None:
return create_single_signal(k1=k1, k2=k2, k3=k3)
ma_key = f"{ma_type}_{timeperiod}"
当前均线 = 当前K线.指标.均线.get(ma_key)
if 当前均线 is None:
return create_single_signal(k1=k1, k2=k2, k3=k3)
当前价 = 当前K线.收盘价
v1 = "看多" if 当前价 > 当前均线 else "看空"
# 均线方向：需要前一根K线的均线值
if len(普K序列) >= di + 2:
前K线 = 普K序列[-di - 1]
if 前K线.指标 is not None:
前均线 = 前K线.指标.均线.get(ma_key)
if 前均线 is not None:
v2 = "向上" if 当前均线 > 前均线 else "向下"
else:
v2 = "任意"
else:
v2 = "任意"
else:
v2 = "任意"
return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2)
def tas_macd_direct_V221106(c, **kwargs) -> OrderedDict:
"""MACD 方向信号 — DIF 在零轴上方为多头，下方为空头
参数模板："{freq}_D{di}#MACD#{fast}#{slow}#{signal}_MACD方向V221106"
**信号逻辑：**
1. DIF > 0，多头；反之，空头
2. DIF 值变化趋势（与前一根比较）：向上/向下
**信号列表：**
- Signal('15分钟_D1#MACD#13#31#11_MACD方向V221106_看多_向上_任意_0')
- Signal('15分钟_D1#MACD#13#31#11_MACD方向V221106_看多_向下_任意_0')
- Signal('15分钟_D1#MACD#13#31#11_MACD方向V221106_看空_向上_任意_0')
- Signal('15分钟_D1#MACD#13#31#11_MACD方向V221106_看空_向下_任意_0')
:param c: 观察者对象
:param kwargs: 其他参数
- fast: 快线周期（默认 13）
- slow: 慢线周期（默认 31）
- signal: 信号周期（默认 11）
- di: 信号计算截止倒数第i根K线
:return: 信号识别结果
"""
fast = int(kwargs.get("fast", 13))
slow = int(kwargs.get("slow", 31))
signal = int(kwargs.get("signal", 11))
di = int(kwargs.get("di", 1))
freq = kwargs.get("freq", "15分钟")
k1, k2, k3 = f"{freq}_D{di}#MACD#{fast}#{slow}#{signal}_MACD方向V221106".split("_", 2)
普K序列 = c.普通K线序列
if len(普K序列) < di + 1:
return create_single_signal(k1=k1, k2=k2, k3=k3)
当前K线 = 普K序列[-di]
cur_macd = 当前K线.指标.macd if 当前K线.指标 else None
if cur_macd is None or cur_macd.DIF is None:
return create_single_signal(k1=k1, k2=k2, k3=k3)
v1 = "看多" if cur_macd.DIF > 0 else "看空"
if len(普K序列) >= di + 2:
前K线 = 普K序列[-di - 1]
prev_macd = 前K线.指标.macd if 前K线.指标 else None
if prev_macd is not None and prev_macd.DIF is not None:
v2 = "向上" if cur_macd.DIF > prev_macd.DIF else "向下"
else:
v2 = "任意"
else:
v2 = "任意"
return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2)
def macd_金叉(观察员: 观察者, **kwargs) -> OrderedDict:
"""MACD 金叉死叉信号 — DIF 与 DEA 的交叉判断
参数模板："{freq}_D{di}#MACD#{fast}#{slow}#{signal}_MACD交叉V260601"
**信号逻辑：**
1. DIF 上穿 DEA（前一根 DIF <= DEA，当前 DIF > DEA）→ 金叉
2. DIF 下穿 DEA（前一根 DIF >= DEA，当前 DIF < DEA）→ 死叉
**信号列表：**
- Signal('15分钟_D1#MACD#13#31#11_MACD交叉V260601_金叉_任意_任意_0')
- Signal('15分钟_D1#MACD#13#31#11_MACD交叉V260601_死叉_任意_任意_0')
:param 观察员: 观察者对象
:param kwargs: 其他参数
- fast: 快线周期（默认 13）
- slow: 慢线周期（默认 31）
- signal: 信号周期（默认 11）
- di: 信号计算截止倒数第i根K线
:return: 信号识别结果
"""
fast = int(kwargs.get("fast", 13))
slow = int(kwargs.get("slow", 31))
signal = int(kwargs.get("signal", 11))
di = int(kwargs.get("di", 1))
freq = kwargs.get("freq", "15分钟")
k1, k2, k3 = f"{freq}_D{di}#MACD#{fast}#{slow}#{signal}_MACD交叉V260601".split("_", 2)
普K序列 = 观察员.普通K线序列
if len(普K序列) < di + 2:
return create_single_signal(k1=k1, k2=k2, k3=k3)
当前K线 = 普K序列[-di]
前K线 = 普K序列[-di - 1]
cur_macd = 当前K线.指标.macd if 当前K线.指标 else None
prev_macd = 前K线.指标.macd if 前K线.指标 else None
if cur_macd is None or prev_macd is None:
return create_single_signal(k1=k1, k2=k2, k3=k3)
if cur_macd.DIF is None or cur_macd.DEA is None:
return create_single_signal(k1=k1, k2=k2, k3=k3)
if prev_macd.DIF is None or prev_macd.DEA is None:
return create_single_signal(k1=k1, k2=k2, k3=k3)
if prev_macd.DIF <= prev_macd.DEA and cur_macd.DIF > cur_macd.DEA:
v1 = "金叉"
elif prev_macd.DIF >= prev_macd.DEA and cur_macd.DIF < cur_macd.DEA:
v1 = "死叉"
else:
v1 = "任意"
return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
# ==============================================================================
# cxt — 缠论形态信号
# ==============================================================================
def cxt_bi_end_V230222(c, **kwargs) -> OrderedDict:
"""当前是最后笔的第几次新低底分型或新高顶分型，用于笔结束辅助
触发条件：新分型
参数模板："{freq}_D1MO{max_overlap}_BE辅助V230222"
**信号逻辑：**
1. 取最后笔及未成笔的分型
2. 当前如果是顶分型，则看当前顶分型是否新高，是第几个新高
3. 当前如果是底分型，则看当前底分型是否新低，是第几个新低
**信号列表：**
- Signal('日线_D1MO3_BE辅助V230222_新低_第2次_任意_0')
- Signal('日线_D1MO3_BE辅助V230222_新高_第2次_任意_0')
- Signal('日线_D1MO3_BE辅助V230222_新低_第3次_任意_0')
:param c: 观察者对象
:param kwargs:
:return: 信号识别结果
"""
max_overlap = int(kwargs.get("max_overlap", 3))
freq = kwargs.get("freq", "日线")
k1, k2, k3 = f"{freq}_D1MO{max_overlap}_BE辅助V230222".split("_", 2)
分型序列 = c.分型序列
笔序列 = c.笔序列
if len(分型序列) < 2 or len(笔序列) < 1:
return create_single_signal(k1=k1, k2=k2, k3=k3)
最后笔 = 笔序列[-1]
当前分型 = 分型序列[-1]
# 找到最后笔的武（终点分型）在分型序列中的位置
try:
笔终点索引 = next(i for i, f in enumerate(分型序列) if f.时间戳 == 最后笔.武.时间戳 and f.结构 == 最后笔.武.结构)
except StopIteration:
return create_single_signal(k1=k1, k2=k2, k3=k3)
# 取笔终点之后的分型（未成笔的分型）
未成笔分型 = 分型序列[笔终点索引 + 1 :]
if len(未成笔分型) < 1:
return create_single_signal(k1=k1, k2=k2, k3=k3)
if 当前分型.结构.value == "顶":
# 统计从笔终点到当前的顶分型新高次数
笔终点顶高 = 最后笔.武.分型特征值
计数 = 0
for f in 未成笔分型:
if f.结构.value == "顶" and f.分型特征值 > 笔终点顶高:
计数 += 1
笔终点顶高 = f.分型特征值
if 计数 > 0 and 当前分型.分型特征值 >= 笔终点顶高:
v1, v2 = "新高", f"第{计数}次"
else:
v1, v2 = "任意", "任意"
elif 当前分型.结构.value == "底":
笔终点底低 = 最后笔.武.分型特征值
计数 = 0
for f in 未成笔分型:
if f.结构.value == "底" and f.分型特征值 < 笔终点底低:
计数 += 1
笔终点底低 = f.分型特征值
if 计数 > 0 and 当前分型.分型特征值 <= 笔终点底低:
v1, v2 = "新低", f"第{计数}次"
else:
v1, v2 = "任意", "任意"
else:
return create_single_signal(k1=k1, k2=k2, k3=k3)
return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2)
def cxt_停顿分型_V230106(c, **kwargs) -> OrderedDict:
"""停顿分型辅助信号 — 结合分型强度和MACD柱子匹配判断
触发条件：新分型
参数模板："{freq}_D{di}停顿分型_BE辅助V230106"
**信号逻辑：**
判断当前分型是否为停顿分型，结合力度和形态给出信号。
停顿分型 = 分型结构为顶/底 + 强度为强/中 + MACD柱子分型匹配。
**信号列表：**
- Signal('1分钟_D0停顿分型_BE辅助V230106_看空_强_任意_0')
- Signal('1分钟_D0停顿分型_BE辅助V230106_看多_强_任意_0')
- Signal('1分钟_D0停顿分型_BE辅助V230106_看空_中_任意_0')
- Signal('1分钟_D0停顿分型_BE辅助V230106_看多_中_任意_0')
:param c: 观察者对象
:param kwargs:
:return: 信号识别结果
"""
di = int(kwargs.get("di", 0))
freq = kwargs.get("freq", "1分钟")
k1, k2, k3 = f"{freq}_D{di}停顿分型_BE辅助V230106".split("_", 2)
分型序列 = c.分型序列
if len(分型序列) < di + 1:
return create_single_signal(k1=k1, k2=k2, k3=k3)
当前分型 = 分型序列[-(di + 1)]
# 只对顶/底分型产出信号
if 当前分型.结构.value not in ("顶", "底"):
return create_single_signal(k1=k1, k2=k2, k3=k3)
v1 = "看空" if 当前分型.结构.value == "顶" else "看多"
v2 = 当前分型.强度()
# 仅强/中分型 + MACD 柱子匹配时认为是有效的停顿分型
if v2 in ("强", "中") and 当前分型.与MACD柱子分型匹配():
pass # 保持 v1, v2
elif v2 in ("强", "中"):
pass # MACD不匹配也产出，但可能被下游过滤
else:
v1, v2 = "任意", "任意"
return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2)
def cxt_中枢第三买卖点_V230602(c, **kwargs) -> OrderedDict:
"""中枢第三买卖点信号——线段中枢的第三类买卖点识别
触发条件：新中枢
参数模板："{freq}_D1MO{max_overlap}_中枢第三买卖点V230602"
**信号逻辑：**
1. 取最后一个中枢，仅处理线段中枢（标识="中枢<线段>"）
2. 判断中枢状态（中枢之上→三买，中枢之下→三卖）
3. 首次穿越0轴：中枢本级第三买卖点后，DIF首次反向穿越0轴并出现对应底/顶分型
4. 中枢段DEA穿越2：第三买卖线段内部DEA双向穿越0轴（上穿+下穿均发生）
**信号列表：**
- Signal('日线_D1MO3_中枢第三买卖点V230602_首次穿越0轴_三买_任意_0')
- Signal('日线_D1MO3_中枢第三买卖点V230602_首次穿越0轴_三卖_任意_0')
- Signal('日线_D1MO3_中枢第三买卖点V230602_中枢段DEA穿越2_三买_任意_0')
- Signal('日线_D1MO3_中枢第三买卖点V230602_中枢段DEA穿越2_三卖_任意_0')
:param c: 观察者对象
:param kwargs:
- max_overlap: 相同信号最大重叠次数
:return: 信号识别结果
"""
max_overlap = int(kwargs.get("max_overlap", 3))
freq = kwargs.get("freq", "日线")
k1, k2, k3 = f"{freq}_D1MO{max_overlap}_中枢第三买卖点V230602".split("_", 2)
中枢序列 = c.中枢序列
if not 中枢序列:
return create_single_signal(k1=k1, k2=k2, k3=k3)
当前中枢 = 中枢序列[-1]
if 当前中枢.标识 != "中枢<线段>":
return create_single_signal(k1=k1, k2=k2, k3=k3)
状态 = 当前中枢.当前状态()
if 状态 == "中枢之中":
return create_single_signal(k1=k1, k2=k2, k3=k3)
if 状态 == "中枢之上":
v2 = "三买"
elif 状态 == "中枢之下":
v2 = "三卖"
else:
return create_single_signal(k1=k1, k2=k2, k3=k3)
v1 = None
# 1. 首次穿越0轴：本级第三买卖点后，DIF反向穿越0轴并出现对应分型
if 当前中枢.本级_第三买卖线 is not None and 当前中枢.完整性("合"):
第三买卖虚线 = 当前中枢.本级_第三买卖线
中K线 = 第三买卖虚线.武.中
缠K序列 = c.缠论K线序列
try:
起点索引 = 缠K序列.index(中K线)
except ValueError:
起点索引 = 0
之后缠K序列 = 缠K序列[起点索引:]
之后缠K = None
if 状态 == "中枢之上" and 中K线.标的K线.macd.DIF > 0:
for k in 之后缠K序列:
if k.标的K线.macd.DIF < 0 and 之后缠K is None:
之后缠K = k
if 之后缠K is not None:
if k.分型 is 分型结构.底 and k.标的K线.macd.DIF < 0:
v1 = "首次穿越0轴"
break
elif 状态 == "中枢之下" and 中K线.标的K线.macd.DIF < 0:
for k in 之后缠K序列:
if k.标的K线.macd.DIF > 0 and 之后缠K is None:
之后缠K = k
if 之后缠K is not None:
if k.分型 is 分型结构.顶 and k.标的K线.macd.DIF > 0:
v1 = "首次穿越0轴"
break
# 2. 中枢段DEA穿越2：第三买卖线段内部DEA双向穿越0轴
if v1 is None and 当前中枢.第三买卖线 is not None and 当前中枢.完整性("合"):
第三线 = 当前中枢.第三买卖线
if 相对方向.分析(当前中枢.高, 当前中枢.低, 第三线.高, 第三线.低).是否缺口():
普K序列 = 第三线.获取普K序列(c.观察员)
MACD特性 = 虚线.统计MACD行为(普K序列, 8, 3)
if MACD特性["DEA上穿0"] > 0 and MACD特性["DEA下穿0"] > 0:
v1 = "中枢段DEA穿越2"
if v1 is None:
return create_single_signal(k1=k1, k2=k2, k3=k3)
return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2)
# ==============================================================================
# bar — K线形态信号
# ==============================================================================
def bar_zdt_V230331(c, **kwargs) -> OrderedDict:
"""计算倒数第di根K线的涨跌停信息
参数模板："{freq}_D{di}_涨跌停V230331"
**信号逻辑：**
- close等于high且大于等于前一根K线的close，近似认为是涨停；反之，跌停。
**信号列表：**
- Signal('15分钟_D1_涨跌停V230331_涨停_任意_任意_0')
- Signal('15分钟_D1_涨跌停V230331_跌停_任意_任意_0')
:param c: 基础周期的观察者对象
:param kwargs:
- di: 倒数第 di 根 K 线
:return: 信号识别结果
"""
di = int(kwargs.get("di", 1))
freq = kwargs.get("freq", "15分钟")
k1, k2, k3 = f"{freq}_D{di}_涨跌停V230331".split("_", 2)
普K序列 = c.普通K线序列
if len(普K序列) < di + 2:
return create_single_signal(k1=k1, k2=k2, k3=k3)
当前K线 = 普K序列[-di]
前K线 = 普K序列[-di - 1]
if 当前K线.收盘价 == 当前K线.高 and 当前K线.收盘价 >= 前K线.收盘价:
v1 = "涨停"
elif 当前K线.收盘价 == 当前K线.低 and 当前K线.收盘价 <= 前K线.收盘价:
v1 = "跌停"
else:
v1 = "任意"
return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
chanlun-py/chanlun/termcolor.py
+296
View File
@@ -0,0 +1,296 @@
# Copyright (c) 2008-2011 Volvox Development Team
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
# Author: Konstantin Lepa <konstantin.lepa@gmail.com>
"""ANSI color formatting for output in terminal."""
from __future__ import annotations
import os
import sys
from functools import cache
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Iterable
from typing import Any
__all__ = ["ATTRIBUTES", "COLORS", "HIGHLIGHTS", "RESET", "can_colorize", "colored", "cprint"]
ATTRIBUTES: dict[str, int] = {
"bold": 1,
"dark": 2,
"italic": 3,
"underline": 4,
"blink": 5,
"reverse": 7,
"concealed": 8,
"strike": 9,
}
HIGHLIGHTS: dict[str, int] = {
"on_black": 40,
"on_grey": 40, # Actually black but kept for backwards compatibility
"on_red": 41,
"on_green": 42,
"on_yellow": 43,
"on_blue": 44,
"on_magenta": 45,
"on_cyan": 46,
"on_light_grey": 47,
"on_dark_grey": 100,
"on_light_red": 101,
"on_light_green": 102,
"on_light_yellow": 103,
"on_light_blue": 104,
"on_light_magenta": 105,
"on_light_cyan": 106,
"on_white": 107,
}
COLORS: dict[str, int] = {
"black": 30,
"grey": 30, # Actually black but kept for backwards compatibility
"red": 31,
"green": 32,
"yellow": 33,
"blue": 34,
"magenta": 35,
"cyan": 36,
"light_grey": 37,
"dark_grey": 90,
"light_red": 91,
"light_green": 92,
"light_yellow": 93,
"light_blue": 94,
"light_magenta": 95,
"light_cyan": 96,
"white": 97,
}
RESET = "\033[0m"
@cache
def can_colorize(*, no_color: bool | None = None, force_color: bool | None = None) -> bool:
"""Check env vars and for tty/dumb terminal"""
# First check overrides:
# "User-level configuration files and per-instance command-line arguments should
# override $NO_COLOR. A user should be able to export $NO_COLOR in their shell
# configuration file as a default, but configure a specific program in its
# configuration file to specifically enable color."
# https://no-color.org
if no_color is not None and no_color:
return False
if force_color is not None and force_color:
return True
# Then check env vars:
if os.environ.get("ANSI_COLORS_DISABLED"):
return False
if os.environ.get("NO_COLOR"):
return False
if os.environ.get("FORCE_COLOR"):
return True
# Then check system:
if os.environ.get("TERM") == "dumb":
return False
if not hasattr(sys.stdout, "fileno"):
return False
try:
return os.isatty(sys.stdout.fileno())
except OSError:
return sys.stdout.isatty()
def _check_rgb(rgb: tuple[int, int, int]) -> None:
if len(rgb) != 3 or not all(0 <= c <= 255 for c in rgb):
msg = f"Expected a tuple of 3 ints in range 0-255, got {rgb!r}"
raise ValueError(msg)
def colored(
text: object,
color: str | tuple[int, int, int] | None = None,
on_color: str | tuple[int, int, int] | None = None,
attrs: Iterable[str] | None = None,
*,
no_color: bool | None = None,
force_color: bool | None = None,
) -> str:
"""Colorize text.
Available text colors:
black, red, green, yellow, blue, magenta, cyan, white,
light_grey, dark_grey, light_red, light_green, light_yellow, light_blue,
light_magenta, light_cyan.
Available text highlights:
on_black, on_red, on_green, on_yellow, on_blue, on_magenta, on_cyan, on_white,
on_light_grey, on_dark_grey, on_light_red, on_light_green, on_light_yellow,
on_light_blue, on_light_magenta, on_light_cyan.
Alternatively, both text colors (color) and highlights (on_color) may
be specified via a tuple of 0-255 ints (R, G, B).
Available attributes:
bold, dark, italic, underline, blink, reverse, concealed, strike.
Example:
colored('Hello, World!', 'red', 'on_black', ['bold', 'blink'])
colored('Hello, World!', 'green')
colored('Hello, World!', (255, 0, 255)) # Purple
"""
result = str(text)
if not can_colorize(no_color=no_color, force_color=force_color):
return result
fmt_str = "\033[%dm%s"
rgb_fore_fmt_str = "\033[38;2;%d;%d;%dm%s"
rgb_back_fmt_str = "\033[48;2;%d;%d;%dm%s"
if color is not None:
if isinstance(color, str):
result = fmt_str % (COLORS[color], result)
elif isinstance(color, tuple):
_check_rgb(color)
result = rgb_fore_fmt_str % (color[0], color[1], color[2], result)
if on_color is not None:
if isinstance(on_color, str):
result = fmt_str % (HIGHLIGHTS[on_color], result)
elif isinstance(on_color, tuple):
_check_rgb(on_color)
result = rgb_back_fmt_str % (on_color[0], on_color[1], on_color[2], result)
if attrs is not None:
for attr in attrs:
result = fmt_str % (ATTRIBUTES[attr], result)
result += RESET
return result
def cprint(
text: object,
color: str | tuple[int, int, int] | None = None,
on_color: str | tuple[int, int, int] | None = None,
attrs: Iterable[str] | None = None,
*,
no_color: bool | None = None,
force_color: bool | None = None,
**kwargs: Any,
) -> None:
"""Print colorized text.
It accepts arguments of print function.
"""
print(
(
colored(
text,
color,
on_color,
attrs,
no_color=no_color,
force_color=force_color,
)
),
**kwargs,
)
if __name__ == "__main__":
print(f"Current terminal type: {os.getenv('TERM')}")
print("Test basic colors:")
cprint("Black color", "black")
cprint("Red color", "red")
cprint("Green color", "green")
cprint("Yellow color", "yellow")
cprint("Blue color", "blue")
cprint("Magenta color", "magenta")
cprint("Cyan color", "cyan")
cprint("White color", "white")
cprint("Light grey color", "light_grey")
cprint("Dark grey color", "dark_grey")
cprint("Light red color", "light_red")
cprint("Light green color", "light_green")
cprint("Light yellow color", "light_yellow")
cprint("Light blue color", "light_blue")
cprint("Light magenta color", "light_magenta")
cprint("Light cyan color", "light_cyan")
print("-" * 78)
print("Test highlights:")
cprint("On black color", on_color="on_black")
cprint("On red color", on_color="on_red")
cprint("On green color", on_color="on_green")
cprint("On yellow color", on_color="on_yellow")
cprint("On blue color", on_color="on_blue")
cprint("On magenta color", on_color="on_magenta")
cprint("On cyan color", on_color="on_cyan")
cprint("On white color", color="black", on_color="on_white")
cprint("On light grey color", on_color="on_light_grey")
cprint("On dark grey color", on_color="on_dark_grey")
cprint("On light red color", on_color="on_light_red")
cprint("On light green color", on_color="on_light_green")
cprint("On light yellow color", on_color="on_light_yellow")
cprint("On light blue color", on_color="on_light_blue")
cprint("On light magenta color", on_color="on_light_magenta")
cprint("On light cyan color", on_color="on_light_cyan")
print("-" * 78)
print("Test attributes:")
cprint("Bold black color", "black", attrs=["bold"])
cprint("Dark red color", "red", attrs=["dark"])
cprint("Italic blue color", "blue", attrs=["italic"])
cprint("Underline green color", "green", attrs=["underline"])
cprint("Blink yellow color", "yellow", attrs=["blink"])
cprint("Reversed blue color", "blue", attrs=["reverse"])
cprint("Concealed magenta color", "magenta", attrs=["concealed"])
cprint("Strike red color", "red", attrs=["strike"])
cprint("Bold underline reverse cyan color", "cyan", attrs=["bold", "underline", "reverse"])
cprint("Dark blink concealed white color", "white", attrs=["dark", "blink", "concealed"])
print("-" * 78)
print("Test mixing:")
cprint("Underline red on black color", "red", "on_black", ["underline"])
cprint("Reversed green on red color", "green", "on_red", ["reverse"])
print("-" * 78)
print("Test RGB:")
cprint("Pure red text (255, 0, 0)", (255, 0, 0))
cprint("Default red for comparison", "red")
cprint("Pure green text (0, 255, 0)", (0, 255, 0))
cprint("Default green for comparison", "green")
cprint("Pure blue text (0, 0, 255)", (0, 0, 255))
cprint("Default blue for comparison", "blue")
cprint("Pure yellow text (255, 255, 0)", (255, 255, 0))
cprint("Default yellow for comparison", "yellow")
cprint("Pure cyan text (0, 255, 255)", (0, 255, 255))
cprint("Default cyan for comparison", "cyan")
cprint("Pure magenta text (255, 0, 255)", (255, 0, 255))
cprint("Default magenta for comparison", "magenta")
cprint("Light pink (255, 182, 193)", (255, 182, 193))
cprint("Light pink (255, 105, 180)", (255, 105, 180))
main.py
+698 -2
View File
@@ -138,6 +138,56 @@ def 收集异常信息(exception: Exception, 上下文: dict = None):
return 错误报告
class 图表展示序列(list):
def __init__(self, 观察员: "观察者"):
super().__init__()
self.观察员 = 观察员
self.序号 = 0
self.__类型标识 = None
def append(self, __object):
if self.序号 > 0:
if __object.标识 != self.__类型标识:
...
self.图表刷新(self[-1], sys._getframe().f_lineno)
else:
self.__类型标识 = __object.标识
super().append(__object)
self.图表添加(__object, sys._getframe().f_lineno)
self.序号 += 1
if __object.标识 in ("线段", "线段<线段>"):
if self.观察员 and self.观察员.配置.线段内部中枢图显:
段: 虚线 = __object
段.合_中枢序列 = 图表展示序列(self.观察员)
段.实_中枢序列 = 图表展示序列(self.观察员)
段.虚_中枢序列 = 图表展示序列(self.观察员)
def pop(self, __index: SupportsIndex = -1):
弹出 = super().pop(__index)
self.图表移除(弹出, sys._getframe().f_lineno)
self.序号 -= 1
return 弹出
def clear(self) -> None:
self.序号 = 0
super().clear()
def 尾部刷新(self, 行号: int):
if self.序号:
self.图表刷新(self[-1], 行号)
def 图表添加(self, 实线: Union["虚线", "中枢"], 行号: int):
self.观察员 and self.观察员.报信(实线, 指令.添加(实线.标识), 行号)
def 图表移除(self, 实线: Union["虚线", "中枢"], 行号: int):
self.观察员 and self.观察员.报信(实线, 指令.删除(实线.标识), 行号)
def 图表刷新(self, 实线: Union["虚线", "中枢"], 行号: int):
self.观察员 and self.观察员.报信(实线, 指令.修改(实线.标识), 行号)
class 时间周期:
def __init__(self, 秒: int, 是否单笔交易: bool = False):
self._秒 = 秒
@@ -452,6 +502,7 @@ class 观察者(观察者):
if 当前买卖点.买卖点K线.时间戳 not in 活跃时间戳序列:
买卖点序列.add(当前买卖点)
当前买卖点.买卖点K线.买卖点信息.add(当前买卖点.备注)
print(当前买卖点, type(当前买卖点))
self.报信(当前买卖点, 指令.添加(当前买卖点.备注), sys._getframe().f_lineno)
def 图表刷新(self):
@@ -721,6 +772,7 @@ class 观察者(观察者):
@classmethod
def 读取数据文件(cls, 文件路径: str, ws=None, 配置=缠论配置()) -> Self:
# btcusd-300-1631772074-1632222374.nb
print(文件路径)
if "_err-" in str(文件路径):
try:
配置 = 缠论配置.加载配置(str(文件路径).replace(".nb", ".json"))
@@ -741,6 +793,320 @@ class 观察者(观察者):
return 实例
def 识别买卖点(self):
"""
简单买卖策略
"""
if not self.笔序列:
return
if self.分型序列[-1].中.序号 + 2 < self.当前缠K.序号:
return
if self.分型序列[-1].强度 not in "强中":
pass
if 笔内部背驰判断(self.普通K线序列, self.笔序列[-1]):
0 and self.添加买卖点("笔", self.笔序列[-1].武, "一", "次次级")
if not self.线段序列:
return
# 观察者.判断线段第二买卖点(self.线段序列[-1], self)
if 笔内部背驰判断(self.普通K线序列, self.线段序列[-1]):
0 and self.添加买卖点("笔", self.线段序列[-1].武, "一", "次级")
if 线段背驰判断(self.普通K线序列, self.线段序列[-1]):
0 and self.添加买卖点("线段", self.线段序列[-1].武, "一", "次级")
if dif_三次穿越背离判断(self.普通K线序列, self.线段序列[-1]):
0 and self.添加买卖点("macd_三次穿越", self.线段序列[-1].武, "一", "次级")
if self.中枢序列:
1 and 观察者.中枢第三买卖点(self.中枢序列[-1], self)
if not self.线段_线段序列:
return
观察者.线段第二买卖点(self.线段_线段序列[-1], self)
观察者.线段第二买卖点(self.线段序列[-1], self)
if 笔内部背驰判断(self.普通K线序列, self.线段_线段序列[-1]):
0 and self.添加买卖点("笔", self.线段_线段序列[-1].武, "一", "本级")
return
@classmethod
def 判断线段第二买卖点(cls, 段: 虚线, 观察员: 观察者):
实, 虚, 第三买卖线, _ = 线段.分割序列(段)
if len(虚) == 2:
符合 = False
# 第一笔 穿越0轴
笔MACD特性 = 虚线.统计MACD行为(虚[0].获取普K序列(观察员), 8, 3)
if 虚[0].方向 is 相对方向.向上 and 笔MACD特性["DEA上穿0"] > 0:
符合 = True
if 虚[0].方向 is 相对方向.向下 and 笔MACD特性["DEA下穿0"] > 0:
符合 = True
# 第二笔 不能穿越0轴
笔MACD特性 = 虚线.统计MACD行为(虚[1].获取普K序列(观察员), 8, 3)
if 虚[1].方向 is 相对方向.向上 and 笔MACD特性["DIF上穿0"] > 0:
符合 = False
if 虚[1].方向 is 相对方向.向下 and 笔MACD特性["DIF下穿0"] > 0:
符合 = False
if 符合:
特征 = "线段二第买卖点"
买卖点分型 = 虚[1].武
虚[1].武.右 and 观察员.添加买卖点(特征, 买卖点分型, "二", "次级")
@classmethod
def 笔中枢当前状态(cls, 当前中枢: "中枢", 观察员: "观察者"):
if 当前中枢.标识 != "中枢<笔>":
return None
普K序列: List[K线] = 观察员.普通K线序列
配置 = 观察员.配置
状态 = 当前中枢.当前状态()
进入段: 虚线 = 观察员.笔序列[观察员.笔序列.index(当前中枢.基础序列[0]) - 1]
离开段: 虚线 = 当前中枢.基础序列[-1]
match 状态:
case "中枢之中":
pass
case "中枢之下" | "中枢之上":
if 进入段.方向 is 离开段.方向 and not 相对方向.分析(进入段.高, 进入段.低, 离开段.高, 离开段.低).是否包含():
if 背驰分析.MACD背驰(进入段, 离开段, 普K序列) or 虚线.买卖意义(离开段, 观察员)[0]:
特征 = "笔中枢"
买卖点分型 = 离开段.武
第几 = "一" if ((离开段.方向 is 相对方向.向上 and 当前中枢.高高 <= 离开段.高) or (离开段.方向 is 相对方向.向下 and 当前中枢.低低 >= 离开段.低)) else "二"
观察员.添加买卖点(特征, 买卖点分型, 第几, "同级")
第三买卖线: 虚线 = 当前中枢.第三买卖线
if 第三买卖线:
买卖点分型 = 当前中枢.第三买卖线.武
同向均值 = 虚线.武之MACD均值_阴(普K序列, 第三买卖线) if 买卖点分型.结构 in (分型结构.底, 分型结构.下) else 虚线.武之MACD均值_阳(普K序列, 第三买卖线)
# if 同向均值 and 第三买卖线.武之MACD均值 and 第三买卖线.武.与MACD柱子匹配 and 第三买卖线.武.与MACD柱子分型匹配:
if 虚线.买卖意义(第三买卖线, 观察员)[0]:
特征 = "笔中枢"
观察员.添加买卖点(特征, 买卖点分型, "三", "同级")
case _:
raise RuntimeError("未知中枢状态", 状态)
@classmethod
def 中枢当前状态(cls, 当前中枢: "中枢", 观察员: "观察者"):
if 当前中枢.标识 != "中枢<线段>":
return None
普K序列: List[K线] = 观察员.普通K线序列
配置 = 观察员.配置
状态 = 当前中枢.当前状态()
买卖点错过误差值 = 配置.买卖点错过误差值
实, 虚, 第三买卖线, _ = 线段.分割序列(当前中枢.基础序列[-1], 当前中枢)
match 状态:
case "中枢之中":
"""if not 虚:
if cls.判断线段内部是否背驰(当前中枢[-1], 观察员) and 当前中枢[-1].武.右 and cls.买卖意义(当前中枢[-1], 观察员)[0]:
特征 = "中枢内背驰"
买卖点分型 = 当前中枢[-1].武
观察员.添加买卖点(特征, 买卖点分型, "一", "次级")"""
case "中枢之下" | "中枢之上":
if 当前中枢.本级_第三买卖线 is not None: # and len(当前中枢) >= 3 and len(虚) >= 2:
买卖点分型 = None
if 当前中枢.完整性("合"):
if 状态 == "中枢之上":
之后缠K序列 = 观察员.缠论K线序列[观察员.缠论K线序列.index(当前中枢.本级_第三买卖线.武.中) :]
之后缠K = None
中枢上轨 = 当前中枢.高
if 当前中枢.本级_第三买卖线.武.中.标的K线.macd.DIF > 0:
for k in 之后缠K序列:
if k.标的K线.macd.DIF < 0: # 首个下穿0轴
if 之后缠K is None:
之后缠K = k
if 之后缠K:
if k.分型 is 分型结构.底 and k.标的K线.macd.DIF < 0:
买卖点分型 = 分型.从缠K序列中获取分型(观察员.缠论K线序列, k)
break
else:
# 中枢之下
之后缠K序列 = 观察员.缠论K线序列[观察员.缠论K线序列.index(当前中枢.本级_第三买卖线.武.中) :]
之后缠K = None
中枢下轨 = 当前中枢.低
if 当前中枢.本级_第三买卖线.武.中.标的K线.macd.DIF < 0:
for k in 之后缠K序列:
if k.标的K线.macd.DIF > 0: # 首个上穿0轴
if 之后缠K is None:
之后缠K = k
if 之后缠K:
if k.分型 is 分型结构.顶 and k.标的K线.macd.DIF > 0:
买卖点分型 = 分型.从缠K序列中获取分型(观察员.缠论K线序列, k)
break
if 买卖点分型:
特征 = "首次穿越0轴"
观察员.添加买卖点(特征, 买卖点分型, "三", "本级")
买卖标的值 = 当前中枢.本级_第三买卖线.武.分型特征值
if 虚线.买卖意义(当前中枢.本级_第三买卖线, 观察员)[0]:
特征 = "中枢段笔"
买卖点分型 = 当前中枢.本级_第三买卖线.武
# 观察员.添加买卖点(特征, 买卖点分型, "三", "本级")
else:
# 错过
for 本级_第三买卖线 in 第三买卖线:
if (当前中枢.本级_第三买卖线 is not 本级_第三买卖线) and 虚线.买卖意义(本级_第三买卖线, 观察员)[0] and (买卖标的值 * (1 + 买卖点错过误差值) > 本级_第三买卖线.武.中.分型特征值 > 买卖标的值 * (1 - 买卖点错过误差值)):
特征 = "中枢段笔"
买卖点分型 = 本级_第三买卖线.武
# 观察员.添加买卖点(特征, 买卖点分型, "三", "错过本级")
break
if 当前中枢.第三买卖线 is None:
# 正在形成第三买卖点？
特征 = 状态
离开段: 虚线 = 当前中枢.基础序列[-1]
"""if not 虚:
# 正在离开中枢
if not 当前中枢.完整性:
pass
else:
pass
if cls.买卖意义(离开段[-1], 观察员)[0] or cls.判断线段内部是否背驰(离开段, 观察员) or cls.买卖意义(离开段[-1], 观察员)[0]:
买卖点分型 = 离开段.武
观察员.添加买卖点(特征, 买卖点分型, "一", "同级")
else:
# 即将到来的 同级第三买卖点！
if (离开段.方向 is 相对方向.向下 and 虚[-1].武.分型特征值 < 离开段.文.分型特征值) or (离开段.方向 is 相对方向.向上 and 虚[-1].武.分型特征值 > 离开段.文.分型特征值):
# 一笔突破当前线段起点
if cls.买卖意义(虚[-1], 观察员)[0]:
买卖点分型 = 虚[-1].武
观察员.添加买卖点(特征, 买卖点分型, "一", "次次级")"""
else:
# 第三买卖点已出现，可能的情况如下
# 1.不当前中枢有任何的交集
# 2.有交集
# 1.与中枢 中高 中低 发生重叠
# 2.与中枢 高高 高低 发生重叠
# 3.失败重新进入当前中枢
# 重点就是第三点，如何判断会回到中枢？
# 提出完整性的概念，最后离开段中内部中枢是否脱离中枢区间
if not 当前中枢.完整性("合"):
return 状态, "不完整"
# if 当前中枢.第三买卖线.武 is not 观察员.分型序列[-1]:
# return 状态, ""
特征 = "中枢段"
assert 相对方向.分析(当前中枢.高, 当前中枢.低, 当前中枢.第三买卖线.高, 当前中枢.第三买卖线.低).是否缺口()
普K序列 = 当前中枢.第三买卖线.获取普K序列(观察员.观察员)
MACD特性 = 虚线.统计MACD行为(普K序列, 8, 3)
if MACD特性["DEA上穿0"] > 0 and MACD特性["DEA下穿0"] > 0:
特征 = "中枢段_DEA穿越2"
买卖点分型 = 当前中枢.第三买卖线.武
观察员.添加买卖点(特征, 买卖点分型, "三", "同级")
if 当前中枢.基础序列[-1].合_中枢序列 and 当前中枢.基础序列[-1].合_中枢序列[-1].基础序列[-1] is 当前中枢.第三买卖线.基础序列[-1]:
pass # return 状态, "前合中枢，线段首次形成"
if 线段.判断线段内部是否背驰(当前中枢.第三买卖线, 观察员): # or 阳[-1]. 买卖意义[0]:
买卖点分型 = 当前中枢.第三买卖线.武
# 观察员.添加买卖点(特征, 买卖点分型, "三", "同级")
else:
return 状态, "同级,内部非背驰"
case _:
raise RuntimeError("未知中枢状态", 状态)
@classmethod
def 中枢第三买卖点(cls, 当前中枢: "中枢", 观察员: "观察者"):
if 当前中枢.标识 != "中枢<线段>":
return None
状态 = 当前中枢.当前状态()
if 状态 == "中枢之中":
return None
实, 虚, 第三买卖线, _ = 线段.分割序列(当前中枢.基础序列[-1], 当前中枢)
if 当前中枢.本级_第三买卖线 is None:
return None
买卖点分型 = None
if 当前中枢.完整性("合"):
之后缠K序列 = 观察员.缠论K线序列[观察员.缠论K线序列.index(当前中枢.本级_第三买卖线.武.中) :]
之后缠K = None
assert 之后缠K序列[0] is 当前中枢.本级_第三买卖线.武.中, (之后缠K序列[0], 当前中枢.本级_第三买卖线.武.中)
if 状态 == "中枢之上":
中枢上轨 = 当前中枢.高
if 当前中枢.本级_第三买卖线.武.中.标的K线.macd.DIF > 0:
for k in 之后缠K序列:
if k.标的K线.macd.DIF < 0: # 首个下穿0轴
if 之后缠K is None:
之后缠K = k
if 之后缠K:
if k.分型 is 分型结构.底 and k.标的K线.macd.DIF < 0:
买卖点分型 = 分型.从缠K序列中获取分型(观察员.缠论K线序列, k)
break
else:
# 中枢之下
中枢下轨 = 当前中枢.低
if 当前中枢.本级_第三买卖线.武.中.标的K线.macd.DIF < 0:
for k in 之后缠K序列:
if k.标的K线.macd.DIF > 0: # 首个上穿0轴
if 之后缠K is None:
之后缠K = k
if 之后缠K:
if k.分型 is 分型结构.顶 and k.标的K线.macd.DIF > 0:
买卖点分型 = 分型.从缠K序列中获取分型(观察员.缠论K线序列, k)
break
if 买卖点分型:
特征 = "首次穿越0轴"
观察员.添加买卖点(特征, 买卖点分型, "三", "本级")
if 当前中枢.第三买卖线 is None:
return None
if not 当前中枢.完整性("合"):
return None
特征 = "中枢段"
assert 相对方向.分析(当前中枢.高, 当前中枢.低, 当前中枢.第三买卖线.高, 当前中枢.第三买卖线.低).是否缺口()
普K序列 = 当前中枢.第三买卖线.获取普K序列(观察员.观察员)
MACD特性 = 虚线.统计MACD行为(普K序列, 8, 3)
if MACD特性["DEA上穿0"] > 0 and MACD特性["DEA下穿0"] > 0:
特征 = "中枢段_DEA穿越2"
买卖点分型 = 当前中枢.第三买卖线.武
观察员.添加买卖点(特征, 买卖点分型, "三", "同级")
@classmethod
def 线段第二买卖点(cls, 当前线段: "虚线", 观察员: "观察者"):
if 当前线段.标识 not in ("线段<线段>", "线段"):
return
实, 虚, _, _ = 线段.分割序列(当前线段, None)
if len(虚) != 2:
return
(首, 尾) = 虚
首_MACD信息 = 虚线.统计MACD行为(首.获取普K序列(观察员))
if 首.方向 is 相对方向.向下 and 首_MACD信息["DEA下穿0"] >= 1:
pass
elif 首.方向 is 相对方向.向上 and 首_MACD信息["DEA上穿0"] >= 1:
pass
else:
return
买卖点分型 = None
尾_MACD信息 = 虚线.统计MACD行为(尾.获取普K序列(观察员))
if 尾.方向 is 相对方向.向下 and 首_MACD信息["DEA下穿0"] >= 1:
买卖点分型 = 尾.武
elif 尾.方向 is 相对方向.向上 and 首_MACD信息["DEA上穿0"] >= 1:
买卖点分型 = 尾.武
else:
return
if not 买卖点分型:
return
特征 = f"{当前线段.标识}第二"
观察员.添加买卖点(特征, 买卖点分型, "二", "同级")
__代码执行器_全局声明__ = dir()
@@ -767,6 +1133,283 @@ def 随机配置(随机源: Optional[random.Random] = None):
)
def dif_三次穿越背离判断(K线序列: List["K线"], 段: 虚线) -> bool:
"""
基于 DIF 线穿越零轴的三次穿越背离判断，第三次穿越后检查 MACD 柱。
向下模式：
1. 第一次穿越：DIF 从正 → 负（下穿0轴）
2. 第二次穿越：DIF 从负 → 正（上穿0轴），且上穿后的最高价 ≤ 第一次下穿前的最高价
3. 第三次穿越：DIF 从正 → 负（再次下穿0轴）
检查第三次穿越后的连续 MACD 负值段内，
价格最低点对应的 MACD 柱值 < 该段 MACD 柱最大值 → 返回 True
向上模式（对称）：
1. 第一次穿越：DIF 从负 → 正（上穿0轴）
2. 第二次穿越：DIF 从正 → 负（下穿0轴），且下穿后的最低价 ≥ 第一次上穿前的最低价
3. 第三次穿越：DIF 从负 → 正（再次上穿0轴）
检查第三次穿越后的连续 MACD 正值段内，
价格最高点对应的 MACD 柱值 < 该段 MACD 柱最大值 → 返回 True
参数:
k线序列: K线对象列表，需包含 .close, .macd.DIF, .macd.MACD柱
方向: "向下" 或 "向上"
返回:
bool: 满足背离条件返回 True，否则 False
"""
方向 = "向上" if 段.方向.是否向上() else "向下"
k线序列: List[K线] = K线.截取(K线序列, 段.文.中.标的K线, 段.武.中.标的K线)
if len(k线序列) < 4:
return False
dif_vals = [k.macd.DEA for k in k线序列]
macd_vals = [k.macd.MACD柱 for k in k线序列]
# 1. 找出所有 DIF 穿越零轴的点
穿越点 = [] # (索引, 方向)
for i in range(1, len(dif_vals)):
prev, curr = dif_vals[i - 1], dif_vals[i]
if prev * curr < 0:
if prev > 0 and curr < 0:
穿越点.append((i, "下穿"))
elif prev < 0 and curr > 0:
穿越点.append((i, "上穿"))
if len(穿越点) < 3:
return False
# 2. 根据方向寻找连续符合顺序的三次穿越
期望序列 = ["下穿", "上穿", "下穿"] if 方向 == "向下" else ["上穿", "下穿", "上穿"]
found = None
for j in range(len(穿越点) - 2):
if 穿越点[j][1] == 期望序列[0] and 穿越点[j + 1][1] == 期望序列[1] and 穿越点[j + 2][1] == 期望序列[2]:
found = (穿越点[j][0], 穿越点[j + 1][0], 穿越点[j + 2][0])
break
if not found:
return False
idx1, idx2, idx3 = found
# 3. 向下模式
if 方向 == "向下":
# 条件2：第二次上穿后的最高价 <= 第一次下穿前的最高价
before = k线序列[:idx1]
if not before:
return False
max_before = max(k.高 for k in before)
after = k线序列[idx2:]
if not after:
return False
max_after = max(k.高 for k in after)
if max_after > max_before:
return False
# 第三次穿越后，取连续 MACD 负值段
segment = []
for i in range(idx3, len(macd_vals)):
if macd_vals[i] < 0:
segment.append(k线序列[i])
else:
break
if len(segment) < 2:
return False
# 段内最低价及其对应的 MACD 柱值
min_price = float("inf")
min_macd = None
for k in segment:
if k.高 < min_price:
min_price = k.高
min_macd = k.macd.MACD柱
max_macd = max(k.macd.MACD柱 for k in segment)
return min_macd < max_macd
# 4. 向上模式
else:
# 条件2：第二次下穿后的最低价 >= 第一次上穿前的最低价
before = k线序列[:idx1]
if not before:
return False
min_before = min(k.低 for k in before)
after = k线序列[idx2:]
if not after:
return False
min_after = min(k.低 for k in after)
if min_after < min_before:
return False
# 第三次穿越后，取连续 MACD 正值段
segment = []
for i in range(idx3, len(macd_vals)):
if macd_vals[i] > 0:
segment.append(k线序列[i])
else:
break
if len(segment) < 2:
return False
# 段内最高价及其对应的 MACD 柱值
max_price = -float("inf")
max_macd = None
for k in segment:
if k.低 > max_price:
max_price = k.低
max_macd = k.macd.MACD柱
max_macd_in_seg = max(k.macd.MACD柱 for k in segment)
return max_macd < max_macd_in_seg
def 找首个MACD交叉前后K线(k线序列: List["K线"], 起始K线: "K线") -> Tuple[Optional["K线"], Optional["K线"]]:
"""
在K线序列中，从起始K线之后查找第一个MACD快慢线交叉点（金叉或死叉），
返回交叉点前一根K线和后一根K线。若未找到，返回(None, None)。
参数:
k线序列: K线对象列表，按时间顺序排列
起始K线: 开始查找的位置
"""
# 定位起始索引
try:
start_idx = k线序列.index(起始K线)
except ValueError:
return None, None
# 从起始K线的下一根开始，到倒数第二根结束（需要比较前后两根）
for i in range(start_idx, len(k线序列) - 1):
prev = k线序列[i]
curr = k线序列[i + 1]
# 获取DIF和DEA值，若存在None则跳过
dif_prev = prev.macd.DIF
dea_prev = prev.macd.DEA
dif_curr = curr.macd.DIF
dea_curr = curr.macd.DEA
if None in (dif_prev, dea_prev, dif_curr, dea_curr):
continue
# 金叉：前一根 DIF <= DEA，后一根 DIF > DEA
if dif_prev <= dea_prev and dif_curr > dea_curr:
return prev, curr
# 死叉：前一根 DIF >= DEA，后一根 DIF < DEA
if dif_prev >= dea_prev and dif_curr < dea_curr:
return prev, curr
return None, None
def 计算MACD柱子分段(k线序列: Sequence["K线"] = None) -> Tuple[List[List["K线"]], ...]:
if not k线序列:
return ()
def 符号(x: float) -> str:
if x > 0:
return "正"
else:
return "负"
当前符号 = 符号(k线序列[0].macd.MACD柱)
当前段柱子 = [k线序列[0]]
结果 = []
for i in range(1, len(k线序列)):
新符号 = 符号(k线序列[i].macd.MACD柱)
if 新符号 == 当前符号:
当前段柱子.append(k线序列[i])
else:
结果.append(当前段柱子)
当前段柱子 = [k线序列[i]]
当前符号 = 新符号
if 当前段柱子:
结果.append(当前段柱子)
正 = []
负 = []
for 序列 in 结果:
if 序列[-1].macd.MACD柱 > 0:
正.append(序列)
else:
负.append(序列)
return 正, 负
def 笔内部背驰判断(K线序列: List[K线], 当前笔: 虚线) -> bool:
"""
基于笔内部MACD柱的分段能量变化，判断是否发生内部背驰（端点可能转折）。
返回 True 表示出现内部背驰信号。
"""
klines: List[K线] = K线.截取(K线序列, 当前笔.文.中.标的K线, 当前笔.武.中.标的K线) # 笔对象本身可迭代返回K线
if len(klines) < 3:
return False
正段, 负段 = 计算MACD柱子分段(klines)
# 按笔的方向选择相关段（向上笔看正段，向下笔看负段）
if 当前笔.方向 == 相对方向.向上:
相关段 = 正段
else:
相关段 = 负段
if len(相关段) < 2:
return False
# 计算每段的能量（代数和）
能量 = [sum(k.macd.MACD柱 for k in seg) for seg in 相关段]
# 计算每段末端价格（向上笔用最高价，向下笔用最低价）
if 当前笔.方向 == 相对方向.向上:
末端价格 = [seg[-1].高 for seg in 相关段]
# 价格必须逐段抬高，能量逐段减小
return all(末端价格[i] < 末端价格[i + 1] for i in range(len(末端价格) - 1)) and all(能量[i] > 能量[i + 1] for i in range(len(能量) - 1))
else:
末端价格 = [seg[-1].低 for seg in 相关段]
# 价格必须逐段降低，能量绝对值逐段减小
return all(末端价格[i] > 末端价格[i + 1] for i in range(len(末端价格) - 1)) and all(abs(能量[i]) > abs(能量[i + 1]) for i in range(len(能量) - 1))
def 线段背驰判断(k线序列: List[K线], 段: 虚线) -> bool:
"""线段内部背驰（比较最后一个中枢的进入段和离开段）"""
if not 段.合_中枢序列:
return False
zs = 段.合_中枢序列[-1] # 最后一个中枢
# 中枢由三笔构成：左、中、右（方向交替）
进入笔 = None
离开笔 = None
# 找到中枢之前的同向笔
for bi in reversed(段.笔序列[: 段.笔序列.index(zs.基础序列[0])]):
if bi.方向 == zs.基础序列[0].方向:
进入笔 = bi
break
# 找到中枢之后的同向笔
for bi in 段.笔序列[段.笔序列.index(zs.基础序列[-1]) + 1 :]:
if bi.方向 == zs.基础序列[0].方向:
离开笔 = bi
break
if 进入笔 is None or 离开笔 is None:
return False
# 价格条件
if zs.基础序列[0].方向 == 相对方向.向上:
if 离开笔.高 <= 进入笔.高:
return False
else:
if 离开笔.低 >= 进入笔.低:
return False
# 力度比较（MACD面积）
return 背驰分析.MACD背驰(离开笔, 进入笔, k线序列)
class 笔K线生成配置(BaseModel):
"""笔的K线生成配置"""
@@ -1255,7 +1898,7 @@ def 从序列中机选(
def 根据当前K线生成新K线(self, 方向: 相对方向, 居中: bool = False) -> "K线":
时间偏移 = timedelta(seconds=self.周期)
时间戳: datetime = self.时间戳 + 时间偏移
时间戳: datetime = self.时间戳 + self.周期 # 时间偏移
成交量: float = 998
高: float = 0
低: float = 0
@@ -1534,6 +2177,7 @@ def 同步_跟踪回测(观察员: 观察者, 数据源: bt.feed.DataBase):
def 测试_读取数据(symbol: str = "btcusd", limit: int = 500, freq: SupportsInt = 时间周期.分(5), ws: Optional[WebSocket] = None, 配置: 缠论配置 = 缠论配置(线段内部中枢图显=False), 文件路径: str = "./templates/btcusd_ex-1800-1685795400-1713488400.nb"):
def 魔法():
启动时间 = datetime.now()
print(观察者)
观察员 = 观察者.读取数据文件(配置.加载文件路径, ws, 配置)
# 观察员.分部分析()
消耗用时 = datetime.now() - 启动时间
@@ -1935,7 +2579,8 @@ async def 处理图表消息(用户标识: str, 消息字典: Dict, websocket: W
config = 消息字典.get("config", dict())
当前配置 = 缠论配置.from_dict(config)
print(当前配置.to_dict())
差异 = 缠论配置().对比(当前配置)
print(差异)
配置组 = 缠论配置.按序号重组字典(当前配置, config)
print(配置组)
@@ -2231,7 +2876,58 @@ async def 主页(
)
def 测试_读取数据2(配置: 缠论配置):
"""测试_读取数据
:param 配置: 缠论配置
:return: 测试函数
"""
def 魔法():
启动时间 = datetime.now()
观察员 = 观察者.读取数据文件(配置.加载文件路径, 配置)
消耗用时 = datetime.now() - 启动时间
print("测试_读取数据 耗时", 消耗用时, "普K数量", len(观察员.普通K线序列))
return 观察员
return 魔法
def 测试_周期合成2(配置: 缠论配置, 配置组: Dict[int, 缠论配置] = dict()):
"""测试_周期合成
:param 配置: 默认配置
:param 配置组: 各周期独立配置
:return: 测试函数
"""
文件路径 = 配置.加载文件路径
name = Path(文件路径).name.split(".")[0]
符号, 周期, 起始时间戳, 结束时间戳 = name.split("-")
周期 = int(周期)
周期组 = [周期, 周期 * 5, 周期 * 5 * 6]
def 魔法():
启动时间 = datetime.now()
多级别分析 = 立体分析器(符号, 周期组, 配置, 配置组)
with open(文件路径, "rb") as f:
buffer = f.read()
size = struct.calcsize(">6d")
for i in range(len(buffer) // size):
k线 = K线.读取大端字节数组(buffer[i * size : i * size + size], 周期, 符号)
多级别分析.投喂K线(k线)
消耗用时 = datetime.now() - 启动时间
print("测试_周期合成", 消耗用时, "普K数量", len(多级别分析._单体分析器[周期].普通K线序列))
return 多级别分析
return 魔法
if __name__ == "__main__":
当前配置 = 缠论配置.不推送()
当前配置.加载文件路径 = str(Path(__file__).parent / "btcusd-300-1761327300-1776327900.nb")
测试_读取数据(配置=当前配置)() # .测试_保存数据()
# 测试_周期合成(当前配置)().测试_保存数据()
if __name__ == "__ma2in__":
def 运行单个回测(线程编号: int):
"""单个线程执行的函数，内部捕获异常以免影响其他线程"""