TacitLab/coinhunter-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: split precheck and smart executor entrypoints

Browse Source
This commit is contained in:
Tacit Lab
2026-04-15 17:03:16 +08:00
parent 6eefabb9ff
commit 965b53b598
10 changed files with 197 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
README.md
View File

@@ -45,3 +45,8 @@ This repository now has a dedicated runtime layer and CLI diagnostics. The next
• services/
• runtime/
• domain logic
The first split is already in place for:
• smart-executor -> commands.smart_executor + services.smart_executor_service
• precheck -> commands.precheck + services.precheck_service

4
src/coinhunter/cli.py
View File

@@ -15,11 +15,11 @@ MODULE_MAP = {
"init": "init_user_state",
"market-probe": "market_probe",
"paths": "paths",
"precheck": "precheck",
"precheck": "commands.precheck",
"review-context": "review_context",
"review-engine": "review_engine",
"rotate-external-gate-log": "rotate_external_gate_log",
"smart-executor": "smart_executor",
"smart-executor": "commands.smart_executor",
"auto-trader": "auto_trader",
}

1
src/coinhunter/commands/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""CLI command adapters for CoinHunter."""

15
src/coinhunter/commands/precheck.py Normal file
View File

@@ -0,0 +1,15 @@
"""CLI adapter for precheck."""
from __future__ import annotations
import sys
from ..services.precheck_service import run
def main() -> int:
return run(sys.argv[1:])
if __name__ == "__main__":
raise SystemExit(main())

15
src/coinhunter/commands/smart_executor.py Normal file
View File

@@ -0,0 +1,15 @@
"""CLI adapter for smart executor."""
from __future__ import annotations
import sys
from ..services.smart_executor_service import run
def main() -> int:
return run(sys.argv[1:])
if __name__ == "__main__":
raise SystemExit(main())

30
src/coinhunter/precheck.py
View File

@@ -924,35 +924,9 @@ def ack_analysis(note: str = ""):
def main():
if len(sys.argv) > 1 and sys.argv[1] == "--ack":
ack_analysis(" ".join(sys.argv[2:]).strip())
return
if len(sys.argv) > 1 and sys.argv[1] == "--mark-run-requested":
mark_run_requested(" ".join(sys.argv[2:]).strip())
return
from .services.precheck_service import run
try:
state = sanitize_state_for_stale_triggers(load_state())
snapshot = build_snapshot()
analysis = analyze_trigger(snapshot, state)
save_state(update_state_after_observation(state, snapshot, analysis))
print(json.dumps(analysis, ensure_ascii=False, indent=2))
except Exception as e:
failure = {
"generated_at": utc_iso(),
"status": "deep_analysis_required",
"should_analyze": True,
"pending_trigger": True,
"cooldown_active": False,
"reasons": ["precheck-error"],
"hard_reasons": ["precheck-error"],
"soft_reasons": [],
"soft_score": 0,
"details": [str(e)],
"compact_summary": f"预检查失败，转入深度分析兜底: {e}",
}
print(json.dumps(failure, ensure_ascii=False, indent=2))
sys.exit(0)
return run(sys.argv[1:])
if __name__ == "__main__":

1
src/coinhunter/services/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Application services for CoinHunter."""

43
src/coinhunter/services/precheck_service.py Normal file
View File

@@ -0,0 +1,43 @@
"""Service entrypoint for precheck workflows."""
from __future__ import annotations
import json
import sys
from .. import precheck as precheck_module
def run(argv: list[str] | None = None) -> int:
argv = list(sys.argv[1:] if argv is None else argv)
if argv and argv[0] == "--ack":
precheck_module.ack_analysis(" ".join(argv[1:]).strip())
return 0
if argv and argv[0] == "--mark-run-requested":
precheck_module.mark_run_requested(" ".join(argv[1:]).strip())
return 0
try:
state = precheck_module.sanitize_state_for_stale_triggers(precheck_module.load_state())
snapshot = precheck_module.build_snapshot()
analysis = precheck_module.analyze_trigger(snapshot, state)
precheck_module.save_state(precheck_module.update_state_after_observation(state, snapshot, analysis))
print(json.dumps(analysis, ensure_ascii=False, indent=2))
return 0
except Exception as exc:
failure = {
"generated_at": precheck_module.utc_iso(),
"status": "deep_analysis_required",
"should_analyze": True,
"pending_trigger": True,
"cooldown_active": False,
"reasons": ["precheck-error"],
"hard_reasons": ["precheck-error"],
"soft_reasons": [],
"soft_score": 0,
"details": [str(exc)],
"compact_summary": f"预检查失败，转入深度分析兜底: {exc}",
}
print(json.dumps(failure, ensure_ascii=False, indent=2))
return 0

111
src/coinhunter/services/smart_executor_service.py Normal file
View File

@@ -0,0 +1,111 @@
"""Service entrypoint for smart executor workflows."""
from __future__ import annotations
import os
import sys
from .. import smart_executor as smart_executor_module
def run(argv: list[str] | None = None) -> int:
argv = list(sys.argv[1:] if argv is None else argv)
args, normalized_argv = smart_executor_module.parse_cli_args(argv)
action = args.command.replace("-", "_")
argv_tail = smart_executor_module.cli_action_args(args, action)
decision_id = (
args.decision_id
or os.getenv("DECISION_ID")
or smart_executor_module.default_decision_id(action, normalized_argv)
)
if args.dry_run:
smart_executor_module.DRY_RUN = True
previous = smart_executor_module.get_execution_state(decision_id)
read_only_action = action in {"balance", "balances", "status"}
if previous and previous.get("status") == "success" and not read_only_action:
smart_executor_module.log(f"⚠️ decision_id={decision_id} 已执行成功，跳过重复执行")
return 0
try:
ex = smart_executor_module.get_exchange()
if read_only_action:
if action in {"balance", "balances"}:
smart_executor_module.command_balances(ex)
else:
smart_executor_module.command_status(ex)
return 0
decision_context = smart_executor_module.build_decision_context(ex, action, argv_tail, decision_id)
if args.analysis:
decision_context["analysis"] = args.analysis
elif os.getenv("DECISION_ANALYSIS"):
decision_context["analysis"] = os.getenv("DECISION_ANALYSIS")
if args.reasoning:
decision_context["reasoning"] = args.reasoning
elif os.getenv("DECISION_REASONING"):
decision_context["reasoning"] = os.getenv("DECISION_REASONING")
smart_executor_module.record_execution_state(
decision_id,
{"status": "pending", "started_at": smart_executor_module.bj_now_iso(), "action": action, "args": argv_tail},
)
if action == "sell_all":
result = smart_executor_module.action_sell_all(ex, args.symbol, decision_id, decision_context)
elif action == "buy":
result = smart_executor_module.action_buy(ex, args.symbol, float(args.amount_usdt), decision_id, decision_context)
elif action == "rebalance":
result = smart_executor_module.action_rebalance(ex, args.from_symbol, args.to_symbol, decision_id, decision_context)
elif action == "hold":
balances = smart_executor_module.fetch_balances(ex)
positions = smart_executor_module.load_positions()
market_snapshot = smart_executor_module.build_market_snapshot(ex)
smart_executor_module.log_decision(
{
**decision_context,
"balances_after": balances,
"positions_after": positions,
"market_snapshot": market_snapshot,
"analysis": decision_context.get("analysis", "hold"),
"reasoning": decision_context.get("reasoning", "hold"),
"execution_result": {"status": "hold"},
}
)
smart_executor_module.log("😴 决策: 持续持有，无操作")
result = {"status": "hold"}
else:
raise RuntimeError(f"未知动作: {action}；请运行 --help 查看正确 CLI 用法")
smart_executor_module.record_execution_state(
decision_id,
{
"status": "success",
"finished_at": smart_executor_module.bj_now_iso(),
"action": action,
"args": argv_tail,
"result": result,
},
)
smart_executor_module.log(f"✅ 执行完成 decision_id={decision_id}")
return 0
except Exception as exc:
smart_executor_module.record_execution_state(
decision_id,
{
"status": "failed",
"finished_at": smart_executor_module.bj_now_iso(),
"action": action,
"args": argv_tail,
"error": str(exc),
},
)
smart_executor_module.log_error(
"smart_executor",
exc,
decision_id=decision_id,
action=action,
args=argv_tail,
)
smart_executor_module.log(f"❌ 执行失败: {exc}")
return 1

67
src/coinhunter/smart_executor.py
View File

@@ -538,72 +538,9 @@ def cli_action_args(args, action: str) -> list[str]:
def main():
global DRY_RUN
args, normalized_argv = parse_cli_args(sys.argv[1:])
action = args.command.replace("-", "_")
argv_tail = cli_action_args(args, action)
decision_id = args.decision_id or os.getenv("DECISION_ID") or default_decision_id(action, normalized_argv)
if args.dry_run:
DRY_RUN = True
from .services.smart_executor_service import run
previous = get_execution_state(decision_id)
read_only_action = action in {"balance", "balances", "status"}
if previous and previous.get("status") == "success" and not read_only_action:
log(f"⚠️ decision_id={decision_id} 已执行成功，跳过重复执行")
return
try:
ex = get_exchange()
if read_only_action:
if action in {"balance", "balances"}:
command_balances(ex)
else:
command_status(ex)
return
decision_context = build_decision_context(ex, action, argv_tail, decision_id)
if args.analysis:
decision_context["analysis"] = args.analysis
elif os.getenv("DECISION_ANALYSIS"):
decision_context["analysis"] = os.getenv("DECISION_ANALYSIS")
if args.reasoning:
decision_context["reasoning"] = args.reasoning
elif os.getenv("DECISION_REASONING"):
decision_context["reasoning"] = os.getenv("DECISION_REASONING")
record_execution_state(decision_id, {"status": "pending", "started_at": bj_now_iso(), "action": action, "args": argv_tail})
if action == "sell_all":
result = action_sell_all(ex, args.symbol, decision_id, decision_context)
elif action == "buy":
result = action_buy(ex, args.symbol, float(args.amount_usdt), decision_id, decision_context)
elif action == "rebalance":
result = action_rebalance(ex, args.from_symbol, args.to_symbol, decision_id, decision_context)
elif action == "hold":
balances = fetch_balances(ex)
positions = load_positions()
market_snapshot = build_market_snapshot(ex)
log_decision({
**decision_context,
"balances_after": balances,
"positions_after": positions,
"market_snapshot": market_snapshot,
"analysis": decision_context.get("analysis", "hold"),
"reasoning": decision_context.get("reasoning", "hold"),
"execution_result": {"status": "hold"},
})
log("😴 决策: 持续持有，无操作")
result = {"status": "hold"}
else:
raise RuntimeError(f"未知动作: {action}；请运行 --help 查看正确 CLI 用法")
record_execution_state(decision_id, {"status": "success", "finished_at": bj_now_iso(), "action": action, "args": argv_tail, "result": result})
log(f"✅ 执行完成 decision_id={decision_id}")
except Exception as e:
record_execution_state(decision_id, {"status": "failed", "finished_at": bj_now_iso(), "action": action, "args": argv_tail, "error": str(e)})
log_error("smart_executor", e, decision_id=decision_id, action=action, args=argv_tail)
log(f"❌ 执行失败: {e}")
sys.exit(1)
return run(sys.argv[1:])
if __name__ == "__main__":