cillin/evotraders
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add restore-mode task launch flow

Browse Source
This commit is contained in:
cillin
2026-03-24 15:27:35 +08:00
parent 6413edf8c9
commit 8d6c3c5647
12 changed files with 572 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
backend/api/runtime.py
View File

@@ -167,6 +167,8 @@ class RuntimeEventsResponse(BaseModel):
class LaunchConfig(BaseModel):
"""Configuration for launching a new trading task."""
launch_mode: str = Field(default="fresh", description="启动形式: fresh, restore")
restore_run_id: Optional[str] = Field(default=None, description="历史任务 run_id，用于恢复启动")
tickers: List[str] = Field(default_factory=list, description="股票池")
schedule_mode: str = Field(default="daily", description="调度模式: daily, interval")
interval_minutes: int = Field(default=60, ge=1, description="间隔分钟数")
@@ -190,6 +192,19 @@ class LaunchResponse(BaseModel):
message: str
class RuntimeHistoryItem(BaseModel):
run_id: str
run_dir: str
updated_at: Optional[str] = None
total_trades: int = 0
total_asset_value: Optional[float] = None
bootstrap: Dict[str, Any] = Field(default_factory=dict)
class RuntimeHistoryResponse(BaseModel):
runs: List[RuntimeHistoryItem]
class StopResponse(BaseModel):
status: str
message: str
@@ -242,6 +257,96 @@ def _get_run_dir(run_id: str) -> Path:
return PROJECT_ROOT / "runs" / run_id
def _load_run_snapshot(run_id: str) -> Dict[str, Any]:
"""Load a specific run snapshot by run_id."""
snapshot_path = _get_run_dir(run_id) / "state" / "runtime_state.json"
if not snapshot_path.exists():
raise HTTPException(status_code=404, detail=f"Run snapshot not found: {run_id}")
return json.loads(snapshot_path.read_text(encoding="utf-8"))
def _copy_path_if_exists(src: Path, dst: Path) -> None:
if not src.exists():
return
if src.is_dir():
shutil.copytree(src, dst, dirs_exist_ok=True)
else:
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
def _restore_run_assets(source_run_id: str, target_run_dir: Path) -> None:
"""Seed a fresh run directory from a historical run snapshot."""
source_run_dir = _get_run_dir(source_run_id)
if not source_run_dir.exists():
raise HTTPException(status_code=404, detail=f"Source run not found: {source_run_id}")
for relative in [
"team_dashboard",
"agents",
"skills",
"memory",
"state/server_state.json",
"state/runtime.db",
"state/research.db",
]:
_copy_path_if_exists(source_run_dir / relative, target_run_dir / relative)
def _list_runs(limit: int = 50) -> list[RuntimeHistoryItem]:
runs_root = PROJECT_ROOT / "runs"
if not runs_root.exists():
return []
items: list[RuntimeHistoryItem] = []
run_dirs = sorted(
[path for path in runs_root.iterdir() if path.is_dir()],
key=lambda path: path.stat().st_mtime,
reverse=True,
)
for run_dir in run_dirs[: max(1, int(limit))]:
run_id = run_dir.name
runtime_state_path = run_dir / "state" / "runtime_state.json"
summary_path = run_dir / "team_dashboard" / "summary.json"
bootstrap: Dict[str, Any] = {}
updated_at: Optional[str] = None
total_trades = 0
total_asset_value: Optional[float] = None
if runtime_state_path.exists():
try:
snapshot = json.loads(runtime_state_path.read_text(encoding="utf-8"))
context = snapshot.get("context") or {}
bootstrap = dict(context.get("bootstrap_values") or {})
updated_at = snapshot.get("events", [{}])[-1].get("timestamp") if snapshot.get("events") else None
except Exception:
bootstrap = {}
if summary_path.exists():
try:
summary = json.loads(summary_path.read_text(encoding="utf-8"))
total_trades = int(summary.get("totalTrades") or 0)
total_asset_value = float(summary.get("totalAssetValue")) if summary.get("totalAssetValue") is not None else None
except Exception:
total_trades = 0
total_asset_value = None
items.append(
RuntimeHistoryItem(
run_id=run_id,
run_dir=str(run_dir),
updated_at=updated_at,
total_trades=total_trades,
total_asset_value=total_asset_value,
bootstrap=bootstrap,
)
)
return items
def _is_timestamped_run_dir(path: Path) -> bool:
try:
datetime.strptime(path.name, "%Y%m%d_%H%M%S")
@@ -390,6 +495,12 @@ async def get_runtime_events() -> RuntimeEventsResponse:
)
@router.get("/history", response_model=RuntimeHistoryResponse)
async def get_runtime_history(limit: int = 20) -> RuntimeHistoryResponse:
"""List recent historical runs for restore/start selection."""
return RuntimeHistoryResponse(runs=_list_runs(limit=limit))
@router.get("/gateway/status", response_model=GatewayStatusResponse)
async def get_gateway_status() -> GatewayStatusResponse:
"""Get Gateway process status and port."""
@@ -609,9 +720,44 @@ async def start_runtime(
_stop_gateway()
await asyncio.sleep(1) # Wait for port release
# 2. Generate run ID and directory
run_id = _generate_run_id()
run_dir = _get_run_dir(run_id)
launch_mode = str(config.launch_mode or "fresh").strip().lower()
if launch_mode not in {"fresh", "restore"}:
raise HTTPException(status_code=400, detail="launch_mode must be 'fresh' or 'restore'")
# 2. Resolve run ID, directory, and bootstrap
if launch_mode == "restore":
restore_run_id = str(config.restore_run_id or "").strip()
if not restore_run_id:
raise HTTPException(status_code=400, detail="restore_run_id is required when launch_mode=restore")
snapshot = _load_run_snapshot(restore_run_id)
context = snapshot.get("context") or {}
if not context.get("config_name"):
raise HTTPException(status_code=404, detail=f"Run context not found: {restore_run_id}")
run_id = restore_run_id
run_dir = _get_run_dir(run_id)
bootstrap = dict(context.get("bootstrap_values") or {})
bootstrap["launch_mode"] = "restore"
bootstrap["restore_run_id"] = restore_run_id
else:
run_id = _generate_run_id()
run_dir = _get_run_dir(run_id)
bootstrap = {
"launch_mode": "fresh",
"restore_run_id": None,
"tickers": config.tickers,
"schedule_mode": config.schedule_mode,
"interval_minutes": config.interval_minutes,
"trigger_time": config.trigger_time,
"max_comm_cycles": config.max_comm_cycles,
"initial_cash": config.initial_cash,
"margin_requirement": config.margin_requirement,
"enable_memory": config.enable_memory,
"mode": config.mode,
"start_date": config.start_date,
"end_date": config.end_date,
"poll_interval": config.poll_interval,
"enable_mock": config.enable_mock,
}
retention_keep = max(1, int(os.getenv("RUNS_RETENTION_COUNT", "20") or "20"))
pruned_run_ids = _prune_old_timestamped_runs(
@@ -621,23 +767,6 @@ async def start_runtime(
if pruned_run_ids:
logger.info("Pruned old run directories: %s", ", ".join(pruned_run_ids))
# 3. Prepare bootstrap config
bootstrap = {
"tickers": config.tickers,
"schedule_mode": config.schedule_mode,
"interval_minutes": config.interval_minutes,
"trigger_time": config.trigger_time,
"max_comm_cycles": config.max_comm_cycles,
"initial_cash": config.initial_cash,
"margin_requirement": config.margin_requirement,
"enable_memory": config.enable_memory,
"mode": config.mode,
"start_date": config.start_date,
"end_date": config.end_date,
"poll_interval": config.poll_interval,
"enable_mock": config.enable_mock,
}
# 4. Create runtime manager
manager = TradingRuntimeManager(
config_name=run_id,

21
backend/apps/news_service.py
View File

@@ -6,15 +6,15 @@ from __future__ import annotations
from typing import Any
from fastapi import Depends, FastAPI, Query
from fastapi.middleware.cors import CORSMiddleware
from backend.apps.cors import add_cors_middleware
from backend.data.market_store import MarketStore
from backend.domains import news as news_domain
def get_market_store() -> MarketStore:
"""Create a market store dependency."""
return MarketStore()
"""Get the MarketStore singleton dependency."""
return MarketStore.get_instance()
def create_app() -> FastAPI:
@@ -25,13 +25,7 @@ def create_app() -> FastAPI:
version="0.1.0",
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
add_cors_middleware(app)
@app.get("/health")
async def health_check() -> dict[str, str]:
@@ -51,6 +45,7 @@ def create_app() -> FastAPI:
start_date=start_date,
end_date=end_date,
limit=limit,
refresh_if_stale=False,
)
@app.get("/api/news-for-date")
@@ -65,6 +60,7 @@ def create_app() -> FastAPI:
ticker=ticker,
date=date,
limit=limit,
refresh_if_stale=False,
)
@app.get("/api/news-timeline")
@@ -79,6 +75,7 @@ def create_app() -> FastAPI:
ticker=ticker,
start_date=start_date,
end_date=end_date,
refresh_if_stale=False,
)
@app.get("/api/categories")
@@ -95,6 +92,7 @@ def create_app() -> FastAPI:
start_date=start_date,
end_date=end_date,
limit=limit,
refresh_if_stale=False,
)
@app.get("/api/similar-days")
@@ -109,6 +107,7 @@ def create_app() -> FastAPI:
ticker=ticker,
date=date,
n_similar=n_similar,
refresh_if_stale=False,
)
@app.get("/api/stories/{ticker}")
@@ -121,6 +120,7 @@ def create_app() -> FastAPI:
store,
ticker=ticker,
as_of_date=as_of_date,
refresh_if_stale=False,
)
@app.get("/api/range-explain")
@@ -139,6 +139,7 @@ def create_app() -> FastAPI:
end_date=end_date,
article_ids=article_ids,
limit=limit,
refresh_if_stale=False,
)
return app

48
backend/domains/news.py
View File

@@ -70,8 +70,14 @@ def get_enriched_news(
start_date: str | None = None,
end_date: str | None = None,
limit: int = 100,
refresh_if_stale: bool = False,
) -> dict[str, Any]:
freshness = ensure_news_fresh(store, ticker=ticker, target_date=end_date)
freshness = ensure_news_fresh(
store,
ticker=ticker,
target_date=end_date,
refresh_if_stale=refresh_if_stale,
)
rows = store.get_news_items_enriched(
ticker,
start_date=start_date,
@@ -101,8 +107,14 @@ def get_news_for_date(
ticker: str,
date: str,
limit: int = 20,
refresh_if_stale: bool = False,
) -> dict[str, Any]:
freshness = ensure_news_fresh(store, ticker=ticker, target_date=date)
freshness = ensure_news_fresh(
store,
ticker=ticker,
target_date=date,
refresh_if_stale=refresh_if_stale,
)
rows = store.get_news_items_enriched(
ticker,
trade_date=date,
@@ -130,8 +142,14 @@ def get_news_timeline(
ticker: str,
start_date: str,
end_date: str,
refresh_if_stale: bool = False,
) -> dict[str, Any]:
freshness = ensure_news_fresh(store, ticker=ticker, target_date=end_date)
freshness = ensure_news_fresh(
store,
ticker=ticker,
target_date=end_date,
refresh_if_stale=refresh_if_stale,
)
timeline = store.get_news_timeline_enriched(
ticker,
start_date=start_date,
@@ -166,8 +184,14 @@ def get_news_categories(
start_date: str | None = None,
end_date: str | None = None,
limit: int = 200,
refresh_if_stale: bool = False,
) -> dict[str, Any]:
freshness = ensure_news_fresh(store, ticker=ticker, target_date=end_date)
freshness = ensure_news_fresh(
store,
ticker=ticker,
target_date=end_date,
refresh_if_stale=refresh_if_stale,
)
rows = store.get_news_items_enriched(
ticker,
start_date=start_date,
@@ -197,8 +221,14 @@ def get_similar_days_payload(
ticker: str,
date: str,
n_similar: int = 5,
refresh_if_stale: bool = False,
) -> dict[str, Any]:
freshness = ensure_news_fresh(store, ticker=ticker, target_date=date)
freshness = ensure_news_fresh(
store,
ticker=ticker,
target_date=date,
refresh_if_stale=refresh_if_stale,
)
result = find_similar_days(
store,
symbol=ticker,
@@ -214,8 +244,14 @@ def get_story_payload(
*,
ticker: str,
as_of_date: str,
refresh_if_stale: bool = False,
) -> dict[str, Any]:
freshness = ensure_news_fresh(store, ticker=ticker, target_date=as_of_date)
freshness = ensure_news_fresh(
store,
ticker=ticker,
target_date=as_of_date,
refresh_if_stale=refresh_if_stale,
)
enrich_news_for_symbol(
store,
ticker,

4
backend/services/gateway_stock_handlers.py
View File

@@ -152,6 +152,7 @@ async def handle_get_stock_news(gateway: Any, websocket: Any, data: dict[str, An
start_date=start_date,
end_date=end_date,
limit=max(limit, 50),
refresh_if_stale=False,
)
news_rows = (payload.get("news") or [])[-limit:]
source = "market_store"
@@ -202,6 +203,7 @@ async def handle_get_stock_news_for_date(gateway: Any, websocket: Any, data: dic
ticker=ticker,
date=trade_date,
limit=limit,
refresh_if_stale=False,
)
news_rows = payload.get("news") or []
source = "market_store"
@@ -255,6 +257,7 @@ async def handle_get_stock_news_timeline(gateway: Any, websocket: Any, data: dic
ticker=ticker,
start_date=start_date,
end_date=end_date,
refresh_if_stale=False,
)
timeline = payload.get("timeline") or []
@@ -313,6 +316,7 @@ async def handle_get_stock_news_categories(gateway: Any, websocket: Any, data: d
start_date=start_date,
end_date=end_date,
limit=200,
refresh_if_stale=False,
)
categories = payload.get("categories") or {}

32
backend/tests/test_news_domain.py
View File

@@ -80,7 +80,7 @@ def test_get_enriched_news_returns_rows_without_enrichment_when_present(monkeypa
monkeypatch.setattr(
news_domain,
"ensure_news_fresh",
lambda store, ticker, target_date=None: {
lambda store, ticker, target_date=None, refresh_if_stale=False: {
"ticker": ticker,
"target_date": target_date,
"last_news_fetch": target_date,
@@ -109,7 +109,7 @@ def test_get_story_and_similar_days_delegate(monkeypatch):
monkeypatch.setattr(
news_domain,
"ensure_news_fresh",
lambda store, ticker, target_date=None: {
lambda store, ticker, target_date=None, refresh_if_stale=False: {
"ticker": ticker,
"target_date": target_date,
"last_news_fetch": target_date,
@@ -137,12 +137,38 @@ def test_get_story_and_similar_days_delegate(monkeypatch):
assert "freshness" in similar
def test_get_enriched_news_defaults_to_read_only_freshness(monkeypatch):
store = _FakeStore()
ensure_calls = []
def fake_ensure(store, ticker, target_date=None, refresh_if_stale=False):
ensure_calls.append(refresh_if_stale)
return {
"ticker": ticker,
"target_date": target_date,
"last_news_fetch": target_date,
"refreshed": False,
}
monkeypatch.setattr(news_domain, "ensure_news_fresh", fake_ensure)
monkeypatch.setattr(news_domain, "news_rows_need_enrichment", lambda rows: False)
payload = news_domain.get_enriched_news(
store,
ticker="AAPL",
end_date="2026-03-16",
)
assert payload["ticker"] == "AAPL"
assert ensure_calls == [False]
def test_get_range_explain_payload_uses_article_ids(monkeypatch):
store = _FakeStore()
monkeypatch.setattr(
news_domain,
"ensure_news_fresh",
lambda store, ticker, target_date=None: {
lambda store, ticker, target_date=None, refresh_if_stale=False: {
"ticker": ticker,
"target_date": target_date,
"last_news_fetch": target_date,

128
backend/tests/test_runtime_service_app.py
View File

@@ -2,6 +2,7 @@
"""Tests for the extracted runtime service app surface."""
import json
from pathlib import Path
from fastapi.testclient import TestClient
@@ -18,6 +19,7 @@ def test_runtime_service_routes_are_exposed():
assert "/api/runtime/start" in paths
assert "/api/runtime/stop" in paths
assert "/api/runtime/cleanup" in paths
assert "/api/runtime/history" in paths
assert "/api/runtime/current" in paths
assert "/api/runtime/gateway/port" in paths
@@ -235,3 +237,129 @@ def test_runtime_cleanup_endpoint_prunes_old_runs(monkeypatch, tmp_path):
assert sorted(payload["pruned_run_ids"]) == ["20260324_090000", "20260324_100000"]
assert (runs_dir / "20260324_110000").exists()
assert (runs_dir / "smoke_fullstack").exists()
def test_runtime_history_lists_recent_runs(monkeypatch, tmp_path):
run_dir = tmp_path / "runs" / "20260324_120000"
(run_dir / "state").mkdir(parents=True)
(run_dir / "team_dashboard").mkdir(parents=True)
(run_dir / "state" / "runtime_state.json").write_text(
json.dumps(
{
"context": {
"config_name": "20260324_120000",
"run_dir": str(run_dir),
"bootstrap_values": {"tickers": ["AAPL"]},
},
"events": [],
}
),
encoding="utf-8",
)
(run_dir / "team_dashboard" / "summary.json").write_text(
json.dumps({"totalTrades": 3, "totalAssetValue": 123456.0}),
encoding="utf-8",
)
monkeypatch.setattr(runtime_module, "PROJECT_ROOT", tmp_path)
with TestClient(create_app()) as client:
response = client.get("/api/runtime/history?limit=5")
assert response.status_code == 200
payload = response.json()
assert payload["runs"][0]["run_id"] == "20260324_120000"
assert payload["runs"][0]["total_trades"] == 3
def test_restore_run_assets_copies_state(monkeypatch, tmp_path):
source_run = tmp_path / "runs" / "20260324_100000"
(source_run / "team_dashboard").mkdir(parents=True)
(source_run / "state").mkdir(parents=True)
(source_run / "agents").mkdir(parents=True)
(source_run / "team_dashboard" / "_internal_state.json").write_text("{}", encoding="utf-8")
(source_run / "state" / "server_state.json").write_text("{}", encoding="utf-8")
target_run = tmp_path / "runs" / "20260324_130000"
monkeypatch.setattr(runtime_module, "PROJECT_ROOT", tmp_path)
runtime_module._restore_run_assets("20260324_100000", target_run)
assert (target_run / "team_dashboard" / "_internal_state.json").exists()
assert (target_run / "state" / "server_state.json").exists()
def test_start_runtime_restore_reuses_historical_run_id(monkeypatch, tmp_path):
run_dir = tmp_path / "runs" / "20260324_100000"
(run_dir / "state").mkdir(parents=True)
(run_dir / "state" / "runtime_state.json").write_text(
json.dumps(
{
"context": {
"config_name": "20260324_100000",
"run_dir": str(run_dir),
"bootstrap_values": {
"tickers": ["AAPL"],
"schedule_mode": "intraday",
"interval_minutes": 30,
"trigger_time": "now",
"max_comm_cycles": 2,
"initial_cash": 100000.0,
"margin_requirement": 0.0,
"enable_memory": False,
"mode": "live",
"poll_interval": 10,
"enable_mock": False,
},
}
}
),
encoding="utf-8",
)
class _DummyManager:
def __init__(self, config_name, run_dir, bootstrap):
self.config_name = config_name
self.run_dir = Path(run_dir)
self.bootstrap = bootstrap
self.context = None
def prepare_run(self):
self.context = type(
"Ctx",
(),
{
"config_name": self.config_name,
"run_dir": self.run_dir,
"bootstrap_values": self.bootstrap,
},
)()
return self.context
class _DummyProcess:
def poll(self):
return None
monkeypatch.setattr(runtime_module, "PROJECT_ROOT", tmp_path)
monkeypatch.setattr(runtime_module, "_find_available_port", lambda start_port=8765, max_port=9000: 8765)
monkeypatch.setattr(runtime_module, "_start_gateway_process", lambda **kwargs: _DummyProcess())
monkeypatch.setattr(runtime_module, "_stop_gateway", lambda: True)
monkeypatch.setattr("backend.runtime.manager.TradingRuntimeManager", _DummyManager)
runtime_state = runtime_module.get_runtime_state()
runtime_state.gateway_process = None
with TestClient(create_app()) as client:
response = client.post(
"/api/runtime/start",
json={
"launch_mode": "restore",
"restore_run_id": "20260324_100000",
"tickers": [],
},
)
assert response.status_code == 200
payload = response.json()
assert payload["run_id"] == "20260324_100000"
assert payload["run_dir"] == str(run_dir)