Tacit Lab
62c40a9776
- Fix TOCTOU race conditions by wrapping read-modify-write cycles under single-file locks in execution_state, portfolio_service, precheck_state, state_manager, and precheck_service. - Add missing test coverage (96 tests total): - test_review_service.py (15 tests) - test_check_api.py (6 tests) - test_external_gate.py main branches (+10 tests) - test_trade_execution.py new commands (+8 tests) - Unify all agent-consumed JSON messages to English. - Config-ize hardcoded values (volume filter, schema_version) via get_user_config with sensible defaults. - Add 1-hour TTL to exchange cache with force_new override. - Add ruff and mypy to dev dependencies; fix all type errors. - Add __all__ declarations to 11 service modules. - Sync README with new commands, config tuning docs, and PyPI badge. - Publish package as coinhunter==1.0.0 on PyPI with MIT license. - Deprecate coinhunter-cli==1.0.1 with runtime warning. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
"""Execution state helpers (decision deduplication, executions.json)."""
|
|
import hashlib
|
|
|
|
__all__ = [
|
|
"default_decision_id",
|
|
"load_executions",
|
|
"save_executions",
|
|
"record_execution_state",
|
|
"get_execution_state",
|
|
]
|
|
|
|
from ..runtime import get_runtime_paths
|
|
from .file_utils import load_json_locked, read_modify_write_json, save_json_locked
|
|
|
|
|
|
def _paths():
|
|
return get_runtime_paths()
|
|
|
|
|
|
def default_decision_id(action: str, argv_tail: list[str]) -> str:
|
|
from datetime import datetime
|
|
|
|
from .trade_common import CST
|
|
|
|
now = datetime.now(CST)
|
|
bucket_min = (now.minute // 15) * 15
|
|
bucket = now.strftime(f"%Y%m%dT%H{bucket_min:02d}")
|
|
raw = f"{bucket}|{action}|{'|'.join(argv_tail)}"
|
|
return hashlib.sha1(raw.encode()).hexdigest()[:16]
|
|
|
|
|
|
def load_executions() -> dict:
|
|
paths = _paths()
|
|
data = load_json_locked(paths.executions_file, paths.executions_lock, {"executions": {}})
|
|
return data.get("executions", {}) # type: ignore[no-any-return]
|
|
|
|
|
|
def save_executions(executions: dict):
|
|
paths = _paths()
|
|
save_json_locked(paths.executions_file, paths.executions_lock, {"executions": executions})
|
|
|
|
|
|
def record_execution_state(decision_id: str, payload: dict):
|
|
paths = _paths()
|
|
read_modify_write_json(
|
|
paths.executions_file,
|
|
paths.executions_lock,
|
|
{"executions": {}},
|
|
lambda data: data.setdefault("executions", {}).__setitem__(decision_id, payload) or data,
|
|
)
|
|
|
|
|
|
def get_execution_state(decision_id: str):
|
|
return load_executions().get(decision_id)
|