diff --git a/README.md b/README.md
index 507e395..60b20ad 100644
--- a/README.md
+++ b/README.md
@@ -47,6 +47,10 @@ predictions/latest_v1/report.json
 
 完整的数据流、参数和目录说明见 [`docs/usage.md`](docs/usage.md)。
 
+## 定时部署
+
+推荐使用常驻 Python 服务：它会在每个完整 8H K 线结束后的宽限期运行一次，使用状态文件防止重启后重复执行，并可将结果推送到 HTTP 服务和 SMTP 邮箱。配置与 systemd 部署方式见 [`docs/deployment.md`](docs/deployment.md)。
+
 ## 手动流程
 
 如果需要单独准备数据和训练模型：
@@ -115,4 +119,3 @@ predictions/latest_v1/report.json
 - [`docs/usage.md`](docs/usage.md)：安装、数据流、命令、目录和故障排查。
 - [`docs/model.md`](docs/model.md)：目标定义、特征、训练协议、指标和限制。
 - [`docs/v0_0823.md`](docs/v0_0823.md)、[`docs/v1_0823.md`](docs/v1_0823.md)：历史实验记录，不是当前运行入口。
-
diff --git a/cbpred/prediction_service.py b/cbpred/prediction_service.py
new file mode 100644
index 0000000..2e30a92
--- /dev/null
+++ b/cbpred/prediction_service.py
@@ -0,0 +1,323 @@
+from __future__ import annotations
+
+import argparse
+import csv
+import fcntl
+import json
+import os
+import re
+import smtplib
+import ssl
+import subprocess
+import sys
+import time
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from email.message import EmailMessage
+from pathlib import Path
+from typing import Any, Iterator
+from urllib.request import Request, urlopen
+
+import yaml
+
+from .pipeline import floor_to_8h
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+ENVIRONMENT_VARIABLE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
+
+
+def _utc_label(timestamp_ms: int) -> str:
+    return datetime.fromtimestamp(timestamp_ms / 1000, timezone.utc).strftime(
+        "%Y-%m-%d %H:%M:%S UTC"
+    )
+
+
+def _expand_environment(value: Any) -> Any:
+    if isinstance(value, str):
+        def replace(match: re.Match[str]) -> str:
+            name = match.group(1)
+            if name not in os.environ:
+                raise ValueError(f"environment variable {name} is required by the configuration")
+            return os.environ[name]
+
+        return ENVIRONMENT_VARIABLE.sub(replace, value)
+    if isinstance(value, list):
+        return [_expand_environment(item) for item in value]
+    if isinstance(value, dict):
+        return {str(key): _expand_environment(item) for key, item in value.items()}
+    return value
+
+
+def _as_mapping(value: Any, name: str) -> dict[str, Any]:
+    if value is None:
+        return {}
+    if not isinstance(value, dict):
+        raise ValueError(f"{name} must be a mapping")
+    return value
+
+
+def load_config(path: Path) -> dict[str, Any]:
+    data = yaml.safe_load(path.read_text()) or {}
+    config = _as_mapping(_expand_environment(data), "root configuration")
+    config["_config_dir"] = path.resolve().parent
+    prediction = _as_mapping(config.get("prediction"), "prediction")
+    if not prediction:
+        raise ValueError("prediction configuration is required")
+    config["prediction"] = prediction
+    config["scheduler"] = _as_mapping(config.get("scheduler"), "scheduler")
+    config["notifications"] = _as_mapping(config.get("notifications"), "notifications")
+    return config
+
+
+def _config_path(config: dict[str, Any], value: str | Path) -> Path:
+    path = Path(value)
+    return path if path.is_absolute() else Path(config["_config_dir"]) / path
+
+
+def _state_path(config: dict[str, Any]) -> Path:
+    value = config["scheduler"].get("state_file", "prediction-service-state.json")
+    return _config_path(config, value)
+
+
+def _read_state(path: Path) -> dict[str, Any]:
+    if not path.exists():
+        return {}
+    data = json.loads(path.read_text())
+    return _as_mapping(data, "service state")
+
+
+def _write_state(path: Path, state: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(state, indent=2, ensure_ascii=False) + "\n")
+    temporary.replace(path)
+
+
+@contextmanager
+def _run_lock(path: Path) -> Iterator[bool]:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a+") as handle:
+        try:
+            fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            yield False
+            return
+        try:
+            yield True
+        finally:
+            fcntl.flock(handle, fcntl.LOCK_UN)
+
+
+def _prediction_command(config: dict[str, Any], end_ms: int) -> tuple[list[str], Path]:
+    prediction = config["prediction"]
+    working_dir = _config_path(config, prediction.get("working_dir", REPO_ROOT)).resolve()
+    command = [sys.executable, "-m", "cbpred.latest_predict", "--end", str(end_ms)]
+    path_options = {
+        "banbot_dir": "--banbot-dir",
+        "banbot_config": "--banbot-config",
+        "data_dir": "--data-dir",
+        "artifact": "--artifact",
+        "output_dir": "--output-dir",
+    }
+    value_options = {
+        "history_months": "--history-months",
+        "holdout_months": "--holdout-months",
+        "min_age_days": "--min-age-days",
+        "chunk_size": "--chunk-size",
+        "dsn": "--dsn",
+        "db_type": "--db-type",
+        "exchange": "--exchange",
+    }
+    for key, option in path_options.items():
+        if key in prediction:
+            command.extend([option, str(_config_path(config, prediction[key]).resolve())])
+    for key, option in value_options.items():
+        if prediction.get(key) is not None:
+            command.extend([option, str(prediction[key])])
+    return command, working_dir
+
+
+def run_prediction(config: dict[str, Any], end_ms: int) -> tuple[dict[str, Any], Path]:
+    command, working_dir = _prediction_command(config, end_ms)
+    print(f"[prediction-service] running completed candle ending {_utc_label(end_ms)}", flush=True)
+    subprocess.run(command, cwd=working_dir, check=True)
+    output_dir = _config_path(
+        config, config["prediction"].get("output_dir", "predictions/latest_v1")
+    ).resolve()
+    report_path = output_dir / "report.json"
+    if not report_path.exists():
+        raise RuntimeError(f"prediction finished without {report_path}")
+    return _as_mapping(json.loads(report_path.read_text()), "prediction report"), output_dir / "predictions.csv"
+
+
+def _prediction_rows(path: Path, max_rows: int | None) -> list[dict[str, str]]:
+    if not path.exists():
+        raise RuntimeError(f"prediction finished without {path}")
+    with path.open(newline="") as handle:
+        rows = list(csv.DictReader(handle))
+    return rows if max_rows is None else rows[:max_rows]
+
+
+def _notification_payload(
+    summary: dict[str, Any],
+    csv_path: Path,
+    end_ms: int,
+    include_predictions: bool,
+    max_rows: int | None,
+) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        "event": "cbpred.prediction.completed",
+        "candle_end_ms": end_ms,
+        "candle_end_utc": _utc_label(end_ms),
+        "summary": summary,
+    }
+    if include_predictions:
+        payload["predictions"] = _prediction_rows(csv_path, max_rows)
+    return payload
+
+
+def send_http_notifications(
+    config: dict[str, Any], summary: dict[str, Any], csv_path: Path, end_ms: int
+) -> int:
+    targets = config["notifications"].get("http", [])
+    if not isinstance(targets, list):
+        raise ValueError("notifications.http must be a list")
+    sent = 0
+    for index, value in enumerate(targets, 1):
+        target = _as_mapping(value, f"notifications.http[{index}]")
+        url = target.get("url")
+        if not isinstance(url, str) or not url:
+            raise ValueError(f"notifications.http[{index}].url is required")
+        max_rows = target.get("max_rows")
+        if max_rows is not None and (not isinstance(max_rows, int) or max_rows < 0):
+            raise ValueError(f"notifications.http[{index}].max_rows must be a non-negative integer")
+        payload = _notification_payload(
+            summary,
+            csv_path,
+            end_ms,
+            bool(target.get("include_predictions", True)),
+            max_rows,
+        )
+        headers = {str(key): str(item) for key, item in _as_mapping(target.get("headers"), "headers").items()}
+        headers.setdefault("Content-Type", "application/json")
+        request = Request(
+            url,
+            data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
+            headers=headers,
+            method=str(target.get("method", "POST")).upper(),
+        )
+        timeout = float(target.get("timeout_seconds", 15))
+        with urlopen(request, timeout=timeout) as response:
+            response.read()
+        sent += 1
+    return sent
+
+
+def send_email_notification(
+    config: dict[str, Any], summary: dict[str, Any], csv_path: Path, end_ms: int
+) -> bool:
+    email_config = config["notifications"].get("email")
+    if not email_config or not _as_mapping(email_config, "notifications.email").get("enabled", True):
+        return False
+    email_config = _as_mapping(email_config, "notifications.email")
+    required = ("host", "from", "to")
+    missing = [name for name in required if not email_config.get(name)]
+    if missing:
+        raise ValueError(f"notifications.email is missing {', '.join(missing)}")
+    recipients = email_config["to"]
+    if not isinstance(recipients, list) or not all(isinstance(item, str) and item for item in recipients):
+        raise ValueError("notifications.email.to must be a non-empty list of addresses")
+
+    subject = str(email_config.get("subject", "cbpred 8H prediction completed"))
+    message = EmailMessage()
+    message["Subject"] = subject
+    message["From"] = str(email_config["from"])
+    message["To"] = ", ".join(recipients)
+    message.set_content(
+        "\n".join(
+            (
+                f"Completed candle: {_utc_label(end_ms)}",
+                f"Latest bar: {summary.get('latest_bar_start_utc')}",
+                f"Predicted symbols: {summary.get('symbols')}",
+                f"Eligible predictions: {summary.get('eligible_predictions')}",
+                f"CSV: {summary.get('csv')}",
+            )
+        )
+        + "\n"
+    )
+    if email_config.get("attach_csv", True):
+        message.add_attachment(csv_path.read_bytes(), maintype="text", subtype="csv", filename=csv_path.name)
+
+    port = int(email_config.get("port", 465 if email_config.get("ssl") else 587))
+    smtp_class = smtplib.SMTP_SSL if email_config.get("ssl") else smtplib.SMTP
+    with smtp_class(str(email_config["host"]), port, timeout=float(email_config.get("timeout_seconds", 30))) as client:
+        if email_config.get("starttls", not email_config.get("ssl")):
+            client.starttls(context=ssl.create_default_context())
+        if email_config.get("username"):
+            client.login(str(email_config["username"]), str(email_config.get("password", "")))
+        client.send_message(message)
+    return True
+
+
+def _ready_end_ms(now_ms: int, grace_seconds: int) -> int | None:
+    end_ms = floor_to_8h(now_ms)
+    return end_ms if now_ms >= end_ms + grace_seconds * 1000 else None
+
+
+def run_ready_cycle(config: dict[str, Any], now_ms: int | None = None) -> str:
+    now_ms = now_ms if now_ms is not None else int(time.time() * 1000)
+    grace_seconds = int(config["scheduler"].get("grace_seconds", 120))
+    if grace_seconds < 0:
+        raise ValueError("scheduler.grace_seconds must be non-negative")
+    end_ms = _ready_end_ms(now_ms, grace_seconds)
+    if end_ms is None:
+        return "waiting for the current 8H candle to close"
+    state_path = _state_path(config)
+    with _run_lock(state_path.with_suffix(state_path.suffix + ".lock")) as acquired:
+        if not acquired:
+            return "another prediction run is in progress"
+        state = _read_state(state_path)
+        if state.get("last_successful_end_ms") == end_ms:
+            return f"completed candle {_utc_label(end_ms)} was already processed"
+        summary, csv_path = run_prediction(config, end_ms)
+        http_count = send_http_notifications(config, summary, csv_path, end_ms)
+        email_sent = send_email_notification(config, summary, csv_path, end_ms)
+        _write_state(
+            state_path,
+            {
+                "last_successful_end_ms": end_ms,
+                "last_successful_end_utc": _utc_label(end_ms),
+                "updated_at_utc": _utc_label(int(time.time() * 1000)),
+            },
+        )
+    return f"processed {_utc_label(end_ms)}; http={http_count}, email={email_sent}"
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Run cbpred after each completed 8H K-line")
+    parser.add_argument("--config", type=Path, required=True, help="YAML service configuration")
+    parser.add_argument("--once", action="store_true", help="check once, then exit")
+    return parser
+
+
+def main() -> None:
+    args = build_parser().parse_args()
+    config = load_config(args.config)
+    interval_seconds = int(config["scheduler"].get("poll_seconds", 60))
+    if interval_seconds <= 0:
+        raise ValueError("scheduler.poll_seconds must be positive")
+    while True:
+        try:
+            print(f"[prediction-service] {run_ready_cycle(config)}", flush=True)
+        except Exception as exc:
+            print(f"[prediction-service] failed: {exc}", file=sys.stderr, flush=True)
+            if args.once:
+                raise
+        if args.once:
+            return
+        time.sleep(interval_seconds)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/config/prediction-service.env.example b/config/prediction-service.env.example
new file mode 100644
index 0000000..cf354a2
--- /dev/null
+++ b/config/prediction-service.env.example
@@ -0,0 +1,4 @@
+# Replace these placeholders in /etc/cbpred/prediction-service.env before the
+# next 8H run. They only allow the example YAML to be parsed.
+CBPRED_WEBHOOK_TOKEN=replace-me
+CBPRED_SMTP_PASSWORD=replace-me
diff --git a/config/prediction-service.user.yml.example b/config/prediction-service.user.yml.example
new file mode 100644
index 0000000..600e56d
--- /dev/null
+++ b/config/prediction-service.user.yml.example
@@ -0,0 +1,42 @@
+# User-systemd variant of prediction-service.yml.example.
+# ${NAME} values are expanded from the user service environment.
+scheduler:
+  poll_seconds: 60
+  grace_seconds: 120
+  state_file: /home/ctyun/.local/state/cbpred/prediction-service-state.json
+
+prediction:
+  working_dir: /data/quant/cbpred
+  banbot_dir: /home/ctyun/ban/banbot
+  banbot_config: /home/ctyun/ban/banbot/biz/config.yml
+  data_dir: /data/quant/cbpred/data
+  artifact: /data/quant/cbpred/artifacts/v1/model.joblib
+  output_dir: /data/quant/cbpred/predictions/latest_v1
+  history_months: 36
+  holdout_months: 10
+  min_age_days: 730
+  chunk_size: 100
+  db_type: auto
+  exchange: binance
+
+notifications:
+  http:
+    - url: https://example.internal/hooks/cbpred
+      method: POST
+      timeout_seconds: 15
+      headers:
+        Authorization: "Bearer ${CBPRED_WEBHOOK_TOKEN}"
+      include_predictions: true
+
+  email:
+    enabled: true
+    host: smtp.example.com
+    port: 465
+    ssl: true
+    username: alerts@example.com
+    password: "${CBPRED_SMTP_PASSWORD}"
+    from: alerts@example.com
+    to:
+      - trader@example.com
+    subject: cbpred 8H prediction completed
+    attach_csv: true
diff --git a/config/prediction-service.yml.example b/config/prediction-service.yml.example
new file mode 100644
index 0000000..75e1f20
--- /dev/null
+++ b/config/prediction-service.yml.example
@@ -0,0 +1,48 @@
+# Copy this file outside the repository, then replace paths and secrets.
+# ${NAME} values are expanded from the service environment; they are never stored in code.
+scheduler:
+  # The daemon polls cheaply, but executes only once for each completed 8H boundary.
+  poll_seconds: 60
+  # Wait for banbot/exchange ingestion after 00:00, 08:00 and 16:00 UTC.
+  grace_seconds: 120
+  state_file: /var/lib/cbpred/prediction-service-state.json
+
+prediction:
+  working_dir: /data/quant/cbpred
+  banbot_dir: /home/ctyun/ban/banbot
+  banbot_config: /home/ctyun/ban/banbot/biz/config.yml
+  data_dir: /data/quant/cbpred/data
+  artifact: /data/quant/cbpred/artifacts/v1/model.joblib
+  output_dir: /data/quant/cbpred/predictions/latest_v1
+  history_months: 36
+  holdout_months: 10
+  min_age_days: 730
+  chunk_size: 100
+  db_type: auto
+  exchange: binance
+
+notifications:
+  http:
+    - url: https://example.internal/hooks/cbpred
+      method: POST
+      timeout_seconds: 15
+      headers:
+        Authorization: "Bearer ${CBPRED_WEBHOOK_TOKEN}"
+      # Set false when the receiving service only needs report.json fields.
+      include_predictions: true
+      # Optional. Without this, all prediction rows are sent.
+      # max_rows: 100
+
+  email:
+    enabled: true
+    host: smtp.example.com
+    port: 465
+    ssl: true
+    # With ssl=true this defaults to false. For port 587, set ssl: false and starttls: true.
+    username: alerts@example.com
+    password: "${CBPRED_SMTP_PASSWORD}"
+    from: alerts@example.com
+    to:
+      - trader@example.com
+    subject: cbpred 8H prediction completed
+    attach_csv: true
diff --git a/docs/deployment.md b/docs/deployment.md
new file mode 100644
index 0000000..21f4c6e
--- /dev/null
+++ b/docs/deployment.md
@@ -0,0 +1,74 @@
+# 8H 预测服务部署
+
+## 方案对比
+
+| 方案 | 适用场景 | 本项目的限制 |
+| --- | --- | --- |
+| cron | 单机、只需固定时间执行一条命令 | 需要额外处理 K 线入库延迟、重启后重复运行、互斥锁、通知失败重试和结构化日志。 |
+| Python 常驻服务 | 需要状态、通知和后续扩展 | 需要由 systemd 或其他进程管理器守护。 |
+
+当前实现选择 Python 常驻服务。它每 `poll_seconds` 秒检查一次，但只会在 UTC 的 `00:00`、`08:00`、`16:00` 后经过 `grace_seconds`，针对刚结束的 8H K 线执行一次。成功状态只会在预测和全部通知都成功后写入；通知或预测失败时，下一次轮询会重试同一根 K 线，进程异常退出时由 systemd 重启。
+
+## 配置
+
+复制 [`config/prediction-service.yml.example`](../config/prediction-service.yml.example) 到 `/etc/cbpred/prediction-service.yml`，填写实际路径、Webhook 和 SMTP 参数。字符串中的 `${NAME}` 会在进程启动时从环境变量读取；缺失的变量会使服务启动失败，避免把空密钥发送到外部服务。
+
+例如，`/etc/cbpred/prediction-service.env`：
+
+```text
+CBPRED_WEBHOOK_TOKEN=replace-with-webhook-token
+CBPRED_SMTP_PASSWORD=replace-with-smtp-password
+```
+
+将其设为仅服务账户可读，例如 `chmod 600 /etc/cbpred/prediction-service.env`。不要把密钥提交到仓库。
+
+HTTP 通知默认发送 `event`、K 线结束时间、`report.json` 摘要和全部 CSV 行；可在各个目标上设置 `include_predictions: false` 或 `max_rows`。邮件会包含摘要并默认附加 `predictions.csv`。HTTP 或邮件任一失败，状态文件不会推进，因此会重试。
+
+## 验证和安装
+
+先执行一次只读调度检查。若当前不在 K 线完成后的宽限期内，它会正常退出且不执行预测：
+
+```bash
+/data/quant/cbpred/.venv/bin/python -m cbpred.prediction_service \
+  --config /etc/cbpred/prediction-service.yml --once
+```
+
+将 [`scripts/cbpred-prediction.service`](../scripts/cbpred-prediction.service) 安装为 systemd unit：
+
+```bash
+sudo install -d -m 755 /etc/cbpred
+sudo install -d -o ctyun -g ctyun -m 750 /var/lib/cbpred
+sudo install -o root -g ctyun -m 640 config/prediction-service.yml.example /etc/cbpred/prediction-service.yml
+sudo install -o root -g ctyun -m 640 config/prediction-service.env.example /etc/cbpred/prediction-service.env
+sudo install -m 644 scripts/cbpred-prediction.service /etc/systemd/system/cbpred-prediction.service
+sudo systemctl daemon-reload
+sudo systemctl enable --now cbpred-prediction.service
+sudo systemctl status cbpred-prediction.service
+```
+
+查看运行日志：
+
+```bash
+journalctl -u cbpred-prediction.service -f
+```
+
+服务使用状态文件锁避免并发重复运行。不要为同一套输出目录同时启用 `update_and_predict.sh` 的 cron 作业和该服务。
+
+没有系统管理员权限时，可使用已启用 linger 的用户级 systemd。它使用同一套服务逻辑，但配置与状态文件放在当前用户目录：
+
+```bash
+install -d -m 700 ~/.config/cbpred ~/.local/state/cbpred ~/.config/systemd/user
+install -m 600 config/prediction-service.user.yml.example ~/.config/cbpred/prediction-service.yml
+install -m 600 config/prediction-service.env.example ~/.config/cbpred/prediction-service.env
+install -m 644 scripts/cbpred-prediction.user.service ~/.config/systemd/user/cbpred-prediction.service
+systemctl --user daemon-reload
+systemctl --user enable --now cbpred-prediction.service
+```
+
+查看用户服务日志：
+
+```bash
+journalctl --user -u cbpred-prediction.service -f
+```
+
+如果只需 cron，可将 `scripts/update_and_predict.sh` 安排在 `2 0,8,16 * * *`（UTC）之后运行，并自行提供锁、通知与失败监控；这正是本服务已经处理的部分。
diff --git a/scripts/cbpred-prediction.service b/scripts/cbpred-prediction.service
new file mode 100644
index 0000000..f6e186c
--- /dev/null
+++ b/scripts/cbpred-prediction.service
@@ -0,0 +1,17 @@
+[Unit]
+Description=cbpred 8H prediction scheduler
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=ctyun
+Group=ctyun
+WorkingDirectory=/data/quant/cbpred
+EnvironmentFile=/etc/cbpred/prediction-service.env
+ExecStart=/data/quant/cbpred/.venv/bin/python -m cbpred.prediction_service --config /etc/cbpred/prediction-service.yml
+Restart=on-failure
+RestartSec=30
+
+[Install]
+WantedBy=multi-user.target
diff --git a/scripts/cbpred-prediction.user.service b/scripts/cbpred-prediction.user.service
new file mode 100644
index 0000000..1a513c2
--- /dev/null
+++ b/scripts/cbpred-prediction.user.service
@@ -0,0 +1,15 @@
+[Unit]
+Description=cbpred 8H prediction scheduler (user)
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+WorkingDirectory=/data/quant/cbpred
+EnvironmentFile=%h/.config/cbpred/prediction-service.env
+ExecStart=/data/quant/cbpred/.venv/bin/python -m cbpred.prediction_service --config %h/.config/cbpred/prediction-service.yml
+Restart=on-failure
+RestartSec=30
+
+[Install]
+WantedBy=default.target
diff --git a/tests/test_prediction_service.py b/tests/test_prediction_service.py
new file mode 100644
index 0000000..8854eb3
--- /dev/null
+++ b/tests/test_prediction_service.py
@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from cbpred.prediction_service import (
+    _prediction_command,
+    _ready_end_ms,
+    load_config,
+    run_ready_cycle,
+    send_email_notification,
+    send_http_notifications,
+)
+
+
+class PredictionServiceTests(unittest.TestCase):
+    def config(self, root: Path) -> dict:
+        return {
+            "_config_dir": root,
+            "scheduler": {"grace_seconds": 120, "state_file": "state.json"},
+            "prediction": {"working_dir": ".", "output_dir": "output"},
+            "notifications": {},
+        }
+
+    def test_load_config_expands_environment_and_requires_prediction(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config_path = root / "service.yml"
+            config_path.write_text("prediction:\n  dsn: '${CBPRED_TEST_DSN}'\n")
+            with patch.dict(os.environ, {"CBPRED_TEST_DSN": "postgres://test"}):
+                config = load_config(config_path)
+            self.assertEqual(config["prediction"]["dsn"], "postgres://test")
+            self.assertEqual(config["_config_dir"], root)
+
+    def test_command_uses_completed_candle_and_resolves_paths(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config = self.config(root)
+            config["prediction"].update({"data_dir": "data", "history_months": 24})
+            command, cwd = _prediction_command(config, 28_800_000)
+            self.assertEqual(cwd, root)
+            self.assertEqual(command[:5], [os.sys.executable, "-m", "cbpred.latest_predict", "--end", "28800000"])
+            self.assertIn(str((root / "data").resolve()), command)
+            self.assertIn("24", command)
+
+    def test_cycle_runs_once_per_completed_candle_and_records_state(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config = self.config(root)
+            csv_path = root / "output" / "predictions.csv"
+            csv_path.parent.mkdir()
+            csv_path.write_text("symbol\nBTC/USDT:USDT\n")
+            summary = {"symbols": 1, "eligible_predictions": 1, "csv": str(csv_path)}
+            completed_end = 28_800_000
+            with patch("cbpred.prediction_service.run_prediction", return_value=(summary, csv_path)) as run:
+                first = run_ready_cycle(config, now_ms=completed_end + 120_000)
+                second = run_ready_cycle(config, now_ms=completed_end + 121_000)
+            run.assert_called_once_with(config, completed_end)
+            self.assertIn("processed", first)
+            self.assertIn("already processed", second)
+            self.assertEqual(json.loads((root / "state.json").read_text())["last_successful_end_ms"], completed_end)
+
+    def test_cycle_waits_before_grace_period(self):
+        self.assertIsNone(_ready_end_ms(28_800_000 + 119_999, 120))
+        self.assertEqual(_ready_end_ms(28_800_000 + 120_000, 120), 28_800_000)
+
+    def test_failed_notification_does_not_advance_state(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config = self.config(root)
+            csv_path = root / "output" / "predictions.csv"
+            csv_path.parent.mkdir()
+            csv_path.write_text("symbol\nBTC/USDT:USDT\n")
+            with patch(
+                "cbpred.prediction_service.run_prediction", return_value=({}, csv_path)
+            ), patch(
+                "cbpred.prediction_service.send_http_notifications", side_effect=RuntimeError("unavailable")
+            ):
+                with self.assertRaisesRegex(RuntimeError, "unavailable"):
+                    run_ready_cycle(config, now_ms=28_920_000)
+            self.assertFalse((root / "state.json").exists())
+
+    def test_http_notification_posts_report_and_predictions(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            csv_path = root / "predictions.csv"
+            csv_path.write_text("symbol,rank\nBTC/USDT:USDT,1\n")
+            config = self.config(root)
+            config["notifications"] = {"http": [{"url": "https://example.test/hook", "headers": {"X-Key": "key"}}]}
+            response = MagicMock()
+            response.__enter__.return_value = response
+            with patch("cbpred.prediction_service.urlopen", return_value=response) as request:
+                count = send_http_notifications(config, {"symbols": 1}, csv_path, 28_800_000)
+            self.assertEqual(count, 1)
+            body = json.loads(request.call_args.args[0].data)
+            self.assertEqual(body["predictions"][0]["symbol"], "BTC/USDT:USDT")
+            self.assertEqual(request.call_args.args[0].headers["Content-type"], "application/json")
+
+    def test_email_notification_uses_ssl_client_and_attaches_csv(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            csv_path = root / "predictions.csv"
+            csv_path.write_text("symbol\nBTC/USDT:USDT\n")
+            config = self.config(root)
+            config["notifications"] = {
+                "email": {
+                    "host": "smtp.example.test", "ssl": True, "username": "user", "password": "secret",
+                    "from": "from@example.test", "to": ["to@example.test"],
+                }
+            }
+            client = MagicMock()
+            client.__enter__.return_value = client
+            with patch("cbpred.prediction_service.smtplib.SMTP_SSL", return_value=client) as smtp:
+                sent = send_email_notification(config, {"symbols": 1}, csv_path, 28_800_000)
+            self.assertTrue(sent)
+            smtp.assert_called_once()
+            client.login.assert_called_once_with("user", "secret")
+            client.send_message.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()