rebuild(stage-11): deliver deterministic heaven workflows

This commit is contained in:
leefer
2026-07-30 07:08:13 +08:00
parent aa3f02bd59
commit 35ae079de7
49 changed files with 7208 additions and 39 deletions
+259
View File
@@ -0,0 +1,259 @@
from __future__ import annotations
from datetime import date, datetime
from typing import Any
from lunar_python import Solar
ELEMENTS = ("", "", "", "", "")
STEM_MOVEMENT = {
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
}
STEM_ELEMENT = {
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
}
BRANCH_ELEMENT = {
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
"": "",
}
SITIAN = {
"": "少阴君火",
"": "少阴君火",
"": "太阴湿土",
"": "太阴湿土",
"": "少阳相火",
"": "少阳相火",
"": "阳明燥金",
"": "阳明燥金",
"": "太阳寒水",
"": "太阳寒水",
"": "厥阴风木",
"": "厥阴风木",
}
ZAIQUAN = {
"少阴君火": "阳明燥金",
"太阴湿土": "太阳寒水",
"少阳相火": "厥阴风木",
"阳明燥金": "少阴君火",
"太阳寒水": "太阴湿土",
"厥阴风木": "少阳相火",
}
QI_SEQUENCE = ("厥阴风木", "少阴君火", "太阴湿土", "少阳相火", "阳明燥金", "太阳寒水")
HOST_SEQUENCE = ("厥阴风木", "少阴君火", "少阳相火", "太阴湿土", "阳明燥金", "太阳寒水")
QI_ELEMENT = {name: name[-1] for name in QI_SEQUENCE}
STEP_NAMES = ("初之气", "二之气", "三之气", "四之气", "五之气", "终之气")
GENERATES = {"": "", "": "", "": "", "": "", "": ""}
CONTROLS = {"": "", "": "", "": "", "": "", "": ""}
CLIMATE = {
"": "风木疏动",
"": "热象渐显",
"": "湿滞偏重",
"": "燥气收敛",
"": "寒意潜行",
}
BEHAVIOR = {
"": ("求新与扩张感增强", "防止把萌芽误作主升", "先写清验证条件"),
"": ("兴奋与急迫感增强", "防止把一致误作确定", "延迟一次冲动决策"),
"": ("对确定性的需求增强", "防止把犹豫误作耐心", "按失效条件做减法"),
"": ("警觉与裁决感增强", "防止过早否定修复", "区分逻辑失效与波动"),
"": ("避险与不确定感增强", "防止放大最坏想象", "降低频率并保留预案"),
}
INDUSTRIES = {
"": ("农业", "林业", "医药", "教育", "纺织", "家居"),
"": ("电力", "新能源", "电子", "半导体", "通信", "传媒"),
"": ("地产", "建筑", "建材", "食品", "零售", "仓储"),
"": ("银行", "证券", "保险", "有色", "机械", "军工"),
"": ("航运", "物流", "水务", "饮料", "化工", "旅游"),
}
def build(trade_date: str, profile: Any | None = None) -> dict[str, Any]:
parsed = date.fromisoformat(trade_date)
solar = Solar.fromYmdHms(parsed.year, parsed.month, parsed.day, 12, 0, 0)
lunar = solar.getLunar()
year_gz = lunar.getYearInGanZhiExact()
month_gz = lunar.getMonthInGanZhiExact()
day_gz = lunar.getDayInGanZhiExact()
sitian = SITIAN[year_gz[1]]
zaiquan = ZAIQUAN[sitian]
step = _qi_step(lunar, solar.toYmd())
host = HOST_SEQUENCE[step - 1]
guest = QI_SEQUENCE[(QI_SEQUENCE.index(sitian) - 2 + step - 1) % 6]
sitian_weight, zaiquan_weight = (15, 5) if step <= 3 else (5, 15)
layers = (
_layer(
"年纲",
(
(STEM_MOVEMENT[year_gz[0]], 30),
(QI_ELEMENT[sitian], sitian_weight),
(QI_ELEMENT[zaiquan], zaiquan_weight),
),
),
_layer("客主加临", ((QI_ELEMENT[host], 20), (QI_ELEMENT[guest], 25))),
_layer("日辰触发", ((STEM_MOVEMENT[day_gz[0]], 2.5), (BRANCH_ELEMENT[day_gz[1]], 2.5))),
)
totals = {element: sum(layer["weights"][element] for layer in layers) for element in ELEMENTS}
balance = sorted(
(
{"element": element, "score": score, "percent": round(score)}
for element, score in totals.items()
),
key=lambda item: item["score"],
reverse=True,
)
primary, secondary = balance[0]["element"], balance[1]["element"]
phrase = f"{CLIMATE[primary]}·{CLIMATE[secondary]}"
behavior = BEHAVIOR[primary]
return {
"date": trade_date,
"lunar_date": f"农历{lunar.getMonthInChinese()}{lunar.getDayInChinese()}",
"pillars": {"year": year_gz, "month": month_gz, "day": day_gz},
"solar_term": {
"current": lunar.getPrevJieQi().getName(),
"next": lunar.getNextJieQi().getName(),
},
"phrase": phrase,
"movement": {
"element": STEM_MOVEMENT[year_gz[0]],
"tendency": "太过" if year_gz[0] in "甲丙戊庚壬" else "不及",
},
"six_qi": {
"sitian": sitian,
"zaiquan": zaiquan,
"step": step,
"step_name": STEP_NAMES[step - 1],
"host": host,
"guest": guest,
},
"layers": [
{
"label": layers[0]["label"],
"dominant": layers[0]["dominant"],
"summary": f"{STEM_MOVEMENT[year_gz[0]]}运为纲,司天{sitian},在泉{zaiquan}",
},
{
"label": layers[1]["label"],
"dominant": layers[1]["dominant"],
"summary": (
f"{guest}加临主{host}{_relation(QI_ELEMENT[guest], QI_ELEMENT[host])}"
),
},
{
"label": layers[2]["label"],
"dominant": layers[2]["dominant"],
"summary": f"{day_gz}日,日干与日支只作轻量触发",
},
],
"balance": balance,
"human_field": {
"emotional_tendency": behavior[0],
"risk": behavior[1],
"balancing_action": behavior[2],
"generation_control": _generation_control(primary, secondary),
},
"personal": _personal(profile, (primary, secondary)),
"sector_catalog": [
{"element": element, "industries": list(INDUSTRIES[element])} for element in ELEMENTS
],
"notice": "五行气场是传统历法与市场行为的象征性观察,不代表可验证的因果关系。",
}
def _personal(profile: Any | None, dominant: tuple[str, str]) -> dict[str, Any] | None:
if profile is None:
return None
born = datetime.strptime(f"{profile.birth_date}T{profile.birth_time}", "%Y-%m-%dT%H:%M")
lunar = Solar.fromYmdHms(born.year, born.month, born.day, born.hour, born.minute, 0).getLunar()
eight = lunar.getEightChar()
pillars = (eight.getYear(), eight.getMonth(), eight.getDay(), eight.getTime())
weights = {element: 0.0 for element in ELEMENTS}
for index, pillar in enumerate(pillars):
weights[STEM_ELEMENT[pillar[0]]] += 1
weights[BRANCH_ELEMENT[pillar[1]]] += 1.5 if index == 1 else 1
day_element = STEM_ELEMENT[eight.getDayGan()]
supportive = {
day_element,
next(element for element, target in GENERATES.items() if target == day_element),
}
hits = [element for element in dominant if element in supportive]
return {
"day_master_element": day_element,
"balance": sorted(weights.items(), key=lambda item: item[1], reverse=True),
"tone": f"当日主气中{''.join(hits)}较合个人生扶倾向"
if hits
else "当日主气与个人生扶倾向交错,宜先察情绪再行动",
"notice": "个人信息只用于本地派生计算,页面不回显出生日期、时辰和性别。",
}
def _layer(label: str, parts: tuple[tuple[str, float], ...]) -> dict[str, Any]:
weights = {element: 0.0 for element in ELEMENTS}
for element, amount in parts:
weights[element] += amount
return {"label": label, "weights": weights, "dominant": max(weights, key=weights.get)}
def _qi_step(lunar: Any, ymd: str) -> int:
current = int(ymd.replace("-", ""))
boundaries = [
int(lunar.getJieQiTable()[name].toYmd().replace("-", ""))
for name in ("大寒", "春分", "小满", "大暑", "秋分", "小雪")
if lunar.getJieQiTable().get(name) is not None
]
if len(boundaries) != 6 or current < boundaries[0] or current >= boundaries[5]:
return 6 if len(boundaries) == 6 else 1
return next(
(index + 1 for index in range(5) if boundaries[index] <= current < boundaries[index + 1]), 6
)
def _relation(guest: str, host: str) -> str:
if guest == host:
return "客主同气"
if GENERATES[guest] == host:
return "客生主,气机相接"
if GENERATES[host] == guest:
return "主生客,时令外泄"
if CONTROLS[guest] == host:
return "客克主,外来变化偏强"
return "主克客,时令与来气相持"
def _generation_control(primary: str, secondary: str) -> str:
if GENERATES[primary] == secondary:
return f"{primary}{secondary},主气向次气流转"
if CONTROLS[primary] == secondary:
return f"{primary}{secondary},主次之气相制"
if GENERATES[secondary] == primary:
return f"{secondary}{primary},次气助主"
if CONTROLS[secondary] == primary:
return f"{secondary}{primary},次气牵制主气"
return f"{primary}{secondary}并见,宜防一端偏盛"