atlas-pattern-generator (0.1.0)
Published 2026-05-01 09:46:49 +08:00 by maode_gitea
Installation
pip install --index-url About this package
Unsupervised online pattern discovery and concept formation system
Atlas PatternGenerator
无监督在线模式发现库。从离散的因子激活信号中自动学习多时间尺度的共现模式，输出可解释的模式文档，支持模型保存与增量训练。
安装
pip install atlas_pattern_generator-0.1.0-py3-none-any.whl
快速开始
from atlas.api import PatternGenerator
pg = PatternGenerator()
# 训练：每 tick 传入一组激活的因子索引
for pattern in stream:
pg.tick(pattern)
# 生命周期管理
pg.weed() # TRA -> SHO 升级
pg.harvest() # SHO -> LON 升级
# 获取已学习模式
patterns = pg.get_patterns()
# 保存模型
pg.save()
PatternGenerator API
引用
from atlas.api import PatternGenerator
构造函数
PatternGenerator(config: Optional[Dict] = None)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
remember_factor |
float |
0.001 |
种子激活时的记忆增量 |
l1_weed_upgrade_threshold |
int |
10 |
L1 层 TRA -> SHO 升级阈值 |
l1_harvest_upgrade_threshold |
int |
20 |
L1 层 SHO -> LON 升级阈值 |
l1_tra_capacity |
int |
100 |
L1 层 TRA 种子容量上限 |
l2_window_size |
int |
4 |
L2 时间窗口（tick 数） |
l2_weed_upgrade_threshold |
int |
15 |
L2 层 TRA -> SHO 升级阈值 |
l2_harvest_upgrade_threshold |
int |
30 |
L2 层 SHO -> LON 升级阈值 |
l2_tra_capacity |
int |
200 |
L2 层 TRA 种子容量上限 |
l2_min_tmp_seed_size |
int |
3 |
L2 最小临时种子大小 |
l3_window_size |
int |
8 |
L3 时间窗口（tick 数） |
l3_weed_upgrade_threshold |
int |
20 |
L3 层 TRA -> SHO 升级阈值 |
l3_harvest_upgrade_threshold |
int |
40 |
L3 层 SHO -> LON 升级阈值 |
l3_tra_capacity |
int |
400 |
L3 层 TRA 种子容量上限 |
l3_min_tmp_seed_size |
int |
3 |
L3 最小临时种子大小 |
enable_subset_suppression |
bool |
False |
是否启用子集抑制 |
enable_layer_subset_suppression |
bool |
False |
是否启用跨层子集抑制 |
min_harvest_purity |
float |
0.0 |
harvest 最小纯度阈值 |
tick(pattern: List[int]) -> None
单步推进。输入当前 tick 的因子激活列表，内部执行 L1 -> L2 -> L3 的完整 grow 流程。
输入格式：List[int]，表示当前 tick 中激活的因子索引。例如 [0, 2, 4] 表示因子 0、2、4 同时激活。
特殊值：空列表 [] 表示当前 tick 无因子激活。
get_patterns(include_sho: bool = False) -> List[int]
获取当前已学习的模式索引列表。
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
include_sho |
bool |
False |
False 仅返回 LON 层（已巩固）模式；True 额外包含 SHO 层（短期）模式 |
返回值：List[int]，模式 ID 列表。L1 模式 ID 范围为 100001+，L2 为 200001+，L3 为 300001+。
weed() -> None
生命周期方法：将各层中达到阈值的 TRA 种子升级为 SHO。
harvest() -> None
生命周期方法：将各层中达到阈值的 SHO 种子升级为 LON，同时清理低质量种子。
export_patterns_doc(factor_names, include_sho) -> Dict[str, Any]
导出所有已发现模式的人类可读文档。
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
factor_names |
Optional[List[str]] |
None |
因子索引到名称的映射，用于生成描述 |
include_sho |
bool |
False |
是否包含 SHO 层模式 |
返回值结构：
{
"summary": {
"total_patterns": int, # 模式总数
"l1_count": int, # L1 模式数
"l2_count": int, # L2 模式数
"l3_count": int, # L3 模式数
"l1_scale": str, # "single-tick co-occurrence"
"l2_scale": str, # 如 "4-tick temporal window"
"l3_scale": str, # 如 "8-tick temporal window"
},
"l1_patterns": [ # L1 模式列表
{
"pattern_id": int,
"level": "L1",
"factor_ids": List[int],
"factor_names": List[str],
"description": str, # 如 "ma5_cross + volume_spike"
"scale": str,
"hit_total": int, # 命中次数
}
],
"l2_patterns": [ # L2 模式列表
{
"pattern_id": int,
"level": "L2",
"composed_of": List[int], # 组成的 L1 模式 ID
"scale": str,
"description": str, # 如 "ma5_cross + volume_spike | rsi_overbought"
"sub_patterns": [...], # 子模式详情
"hit_total": int,
}
],
"l3_patterns": [...], # L3 模式列表，结构与 L2 类似
}
save(path: str = "data/model.json") -> None
保存模型状态到 JSON 文件。自动创建目录。
load(path: str = "data/model.json") -> PatternGenerator
从 JSON 文件加载模型状态。加载后可继续在新数据上训练。
输入输出格式规范
输入：tick(pattern)
# 每 tick 传入一组激活的因子索引
pg.tick([0, 2, 4]) # 因子 0、2、4 同时激活
pg.tick([]) # 无激活
pg.tick([1]) # 仅因子 1 激活
因子索引是连续的整数，从 0 开始。例如有 11 个因子，则索引范围是 0~10。
输出：get_patterns()
patterns = pg.get_patterns()
# 返回: [100001, 100003, 200001]
# 100001 = L1 层模式, 200001 = L2 层模式
输出：export_patterns_doc()
doc = pg.export_patterns_doc(factor_names=["f0", "f1", "f2"])
# 返回结构化字典，包含每层模式的描述、组成、命中次数
完整示例
from atlas.api import PatternGenerator
import json
# 1. 创建实例
pg = PatternGenerator(config={
"l1_weed_upgrade_threshold": 10,
"l1_harvest_upgrade_threshold": 20,
})
# 2. 训练数据：模拟周期性模式
data = [
[0, 1], [0, 1], [0, 1], [],
[0, 1], [0, 1], [0, 1], [],
] * 20 # 重复 20 轮
for pattern in data:
pg.tick(pattern)
# 3. 生命周期管理
pg.weed()
pg.harvest()
# 4. 获取模式
patterns = pg.get_patterns()
print(f"learned {len(patterns)} patterns: {patterns}")
# 5. 导出模式文档
doc = pg.export_patterns_doc(factor_names=["ma5", "ma10", "volume", "rsi"])
with open("data/patterns.json", "w") as f:
json.dump(doc, f, indent=2, ensure_ascii=False)
# 6. 保存模型
pg.save("data/model.json")
# 7. 加载并继续训练
pg2 = PatternGenerator.load("data/model.json")
pg2.tick([2, 3])
架构简述
Atlas 采用三层时间尺度分层架构：
- L1（2 帧窗口）：处理最细粒度的因子共现模式，识别单个 tick 内哪些因子经常一起激活
- L2（4 帧窗口）：处理 L1 输出，识别中期时间尺度的模式组合
- L3（8 帧窗口）：处理 L2 输出，识别长期时间尺度的模式组合
每层拥有独立的 TRA（临时）-> SHO（短期）-> LON（长期）三层记忆结构，通过 weed（轻清理）和 harvest（全重组）实现种子的筛选、巩固与遗忘。
本地打包
构建 pip-installable wheel 包，用于内部制品库分发。
前置依赖
pip install build
构建命令
python -m build --wheel
构建产物位于 dist/ 目录，例如：
dist/atlas_pattern_generator-0.1.0-py3-none-any.whl
本地安装验证
python -m venv /tmp/test_env
/tmp/test_env/bin/pip install dist/atlas_pattern_generator-0.1.0-py3-none-any.whl
/tmp/test_env/bin/python -c "from atlas.api import PatternGenerator; pg = PatternGenerator(); print('OK')"
注意事项
- wheel 包构建到
dist/即完成，不上传到 PyPI，由用户自行上传内部制品库 - 安装后配置文件
main.yaml自动内嵌在site-packages/atlas/configs/下，无需额外配置 - 导出文件默认保存到
data/目录，该目录已被.gitignore忽略
Requirements
Requires Python: >=3.10