scripts/artifacts.py - master - roman.timokhin/traiding-brain

scripts/artifacts.py
715 строк27 KB
roman.timokhin
Тестирование и отладка
30 май 2026, 10:25
30 май 2026, 10:25a97a82f
"""
Модуль работы с артефактами торгового цикла.

Функции:
    - save_evening_scan() — сохранение вечернего артефакта (JSON)
    - save_morning_confirmation() — сохранение утреннего подтверждения
    - save_trade_record() — создание/обновление артефакта сделки
    - update_trade_entry() — запись параметров входа в сделку
    - update_trade_risk_guard() — запись резолюции risk-guard
    - update_trade_close() — закрытие сделки с расчётом P&L
    - load_evening_scan() — загрузка вечернего артефакта
    - load_trades() — загрузка истории сделок
    - load_latest_artifact() — загрузка последнего артефакта по типу
"""
import json
import re
from datetime import datetime, date
from pathlib import Path
from typing import Optional

import numpy as np
import pandas as pd

ROOT_DIR = Path(__file__).resolve().parent.parent
ARTIFACTS_DIR = ROOT_DIR / "data" / "artifacts"
TRADES_DIR = ROOT_DIR / "data" / "trades"

# Убеждаемся, что директории существуют
ARTIFACTS_DIR.mkdir(parents=True, exist_ok=True)
TRADES_DIR.mkdir(parents=True, exist_ok=True)


# ---------------------------------------------------------------------------
# Вспомогательные функции
# ---------------------------------------------------------------------------

def _today_str() -> str:
    """Возвращает сегодняшнюю дату в формате YYYY-MM-DD."""
    return date.today().isoformat()


def _now_str() -> str:
    """Возвращает текущее время в формате YYYY-MM-DD HH:MM."""
    return datetime.now().strftime("%Y-%m-%d %H:%M")


def _signal_text(signal: int) -> str:
    """Преобразует числовой сигнал в текстовое представление.

    Args:
        signal: -1 (SHORT), 0 (---), 1 (LONG)

    Returns:
        Текстовое представление сигнала
    """
    if signal == 1:
        return "LONG"
    elif signal == -1:
        return "SHORT"
    return "---"


# ---------------------------------------------------------------------------
# Сохранение артефактов
# ---------------------------------------------------------------------------

def _save_json_artifact(data: dict, filename: str) -> str:
    """Сохраняет структурированные данные артефакта в JSON.

    Args:
        data: Словарь с данными
        filename: Имя файла (без расширения)

    Returns:
        Путь к сохранённому JSON-файлу
    """
    json_path = ARTIFACTS_DIR / f"{filename}.json"
    json_path.write_text(
        json.dumps(data, indent=2, ensure_ascii=False, default=str),
        encoding="utf-8",
    )
    return str(json_path)


def save_evening_scan(results: list, target_date: Optional[str] = None) -> str:
    """Сохраняет структурированные данные вечернего скрининга в JSON.

    Markdown-отчёт не генерируется — его формирует AI-агент
    на основе JSON и шаблона templates/evening_scan_template.md.

    Args:
        results: Список словарей с результатами анализа тикеров
        target_date: Дата торговой сессии (завтра). По умолчанию today+1.

    Returns:
        Путь к сохранённому JSON-файлу
    """
    if target_date is None:
        from datetime import timedelta
        target_date = (date.today() + timedelta(days=1)).isoformat()

    now = _now_str()
    today = _today_str()

    # Сериализуем сигналы, заменяя NaN/NaT на None для совместимости с JSON
    def _safe_val(v):
        if isinstance(v, float) and (pd.isna(v) or np.isnan(v)):
            return None
        if v is None or v == "" or (isinstance(v, float) and np.isnan(v)):
            return None
        return v

    json_data = {
        "artifact_type": "evening_scan",
        "created": now,
        "target_date": target_date,
        "signals": [
            {
                "ticker": r["ticker"],
                "close": _safe_val(r.get("close")),
                "signal": int(r.get("signal", 0)) if pd.notna(r.get("signal")) else 0,
                "votes": int(r.get("votes", 0)) if pd.notna(r.get("votes")) else 0,
                "rsi": round(r["rsi"], 1) if pd.notna(r.get("rsi")) else None,
                "macd_hist": round(r["macd_hist"], 4) if pd.notna(r.get("macd_hist")) else None,
                "macd_trend": r.get("macd_trend"),
                "bb_pos": round(r["bb_pos"], 2) if pd.notna(r.get("bb_pos")) else None,
                "volume_ratio": round(r["volume_ratio"], 2) if pd.notna(r.get("volume_ratio")) else None,
                "sma_trend": r.get("sma_trend"),
                "entry": round(r["entry"], 2) if pd.notna(r.get("entry")) else None,
                "sl": round(r["sl"], 2) if pd.notna(r.get("sl")) else None,
                "tp": round(r["tp"], 2) if pd.notna(r.get("tp")) else None,
                "risk": round(r["risk"], 1) if pd.notna(r.get("risk")) else None,
                "reward": round(r["reward"], 1) if pd.notna(r.get("reward")) else None,
                "signals_detail": {
                    k: int(v) if pd.notna(v) else 0
                    for k, v in (r.get("signals_detail") or {}).items()
                },
                "corporate_events": r.get("corporate_events", "") or "",
                "corporate_alert": int(r.get("corporate_alert", 0)) if pd.notna(r.get("corporate_alert")) else 0,
            }
            for r in results if r is not None
        ],
    }

    json_path = _save_json_artifact(json_data, f"evening_scan_{today}")
    return str(json_path)


def save_morning_confirmation(confirmations: list, plan_date: Optional[str] = None) -> str:
    """Сохраняет утреннее подтверждение сигналов.

    Args:
        confirmations: Список словарей с результатами подтверждения:
            [
                {
                    "ticker": "VTBR",
                    "planned_signal": -1,
                    "planned_entry": 4883.50,
                    "current_price": 4850.00,
                    "gap_pct": -0.69,
                    "volume_30m_ratio": 1.5,
                    "verdict": "входить",
                    "reason": "Гэпа нет, объём подтверждает"
                },
                ...
            ]
        plan_date: Дата торгового плана (сегодня). По умолчанию today.

    Returns:
        Путь к сохранённому файлу
    """
    if plan_date is None:
        plan_date = _today_str()

    now = _now_str()
    today = _today_str()

    lines = []
    lines.append(f"# Утреннее подтверждение — {today}")
    lines.append(f"**Сформирован:** {now}")
    lines.append(f"**Торговый план от:** evening_scan_{plan_date}.json")
    lines.append("")
    lines.append("---")
    lines.append("")

    lines.append("## Статус сигналов")
    lines.append("")
    lines.append("| Тикер | План | Цена сейчас | Гэп % | Объём 30м | Вердикт |")
    lines.append("|-------|------|-------------|-------|-----------|---------|")

    for c in confirmations:
        ticker = c["ticker"]
        plan_sig = _signal_text(c.get("planned_signal", 0))
        price = c.get("current_price", 0)
        gap = c.get("gap_pct", 0)
        vol = c.get("volume_30m_ratio", None)
        verdict = c.get("verdict", "пропустить")

        vol_s = f"{vol:.2f}x" if vol else "N/A"
        gap_s = f"{gap:+.2f}%" if gap else "N/A"

        lines.append(f"| **{ticker}** | {plan_sig} | {price:.2f} | {gap_s} | {vol_s} | {verdict} |")

    lines.append("")
    lines.append("---")
    lines.append("")

    # --- Рекомендации ---
    to_enter = [c for c in confirmations if c.get("verdict") == "входить"]
    to_skip = [c for c in confirmations if c.get("verdict") != "входить"]

    if to_enter:
        lines.append("## Рекомендации на сегодня")
        lines.append("")
        lines.append("### Входить")
        lines.append("")
        for c in to_enter:
            lines.append(f"- **{c['ticker']}** — {c.get('current_price', 0):.2f} RUB")
            lines.append(f"  - Причина: {c.get('reason', 'N/A')}")
            lines.append("")

    if to_skip:
        lines.append("### Пропустить")
        lines.append("")
        for c in to_skip:
            lines.append(f"- **{c['ticker']}** — {c.get('reason', 'N/A')}")
        lines.append("")

    lines.append("---")
    lines.append("")
    lines.append("*Не является инвестиционной рекомендацией.*")

    content = "\n".join(lines)
    filename = f"morning_confirmation_{today}.md"
    filepath = ARTIFACTS_DIR / filename
    filepath.write_text(content, encoding="utf-8")

    # Сохраняем структурированные данные в JSON для надёжного парсинга
    json_data = {
        "artifact_type": "morning_confirmation",
        "created": now,
        "plan_date": plan_date,
        "confirmations": confirmations,
    }
    _save_json_artifact(json_data, f"morning_confirmation_{today}")

    return str(filepath)


# ---------------------------------------------------------------------------
# Структура артефакта сделки
# ---------------------------------------------------------------------------

def _new_trade_artifact(ticker: str, direction: str) -> dict:
    """Создаёт новый пустой артефакт сделки с секционной структурой.

    Args:
        ticker: Тикер инструмента.
        direction: "LONG" или "SHORT".

    Returns:
        Словарь с полной структурой артефакта сделки.
    """
    date_str = _today_str()
    filename = f"trade_{date_str}_{ticker}.json"

    return {
        "ticker": ticker.upper(),
        "direction": direction.upper(),

        # --- Параметры входа (заполняются при entry) ---
        "entry": {
            "time": None,
            "price": None,
            "volume": 0,
            "value": 0.0,
            "deposit_at_entry": None,
            "signal_type": None,
            "signal_strength": None,
            "total_votes": None,
            "rationale": None,
            "morning_verdict": None,
        },

        # --- Уровни SL/TP (из вечернего скрининга) ---
        "levels": {
            "stop_loss": None,
            "take_profit": None,
            "risk_pct": None,
            "reward_pct": None,
            "risk_reward_ratio": None,
        },

        # --- Расчёт размера позиции (из PortfolioManager) ---
        "position_sizing": {
            "risk_per_trade_pct": None,
            "max_position_pct": None,
            "position_pct": None,
            "risk_amount_rub": None,
            "recommended_volume": 0,
        },

        # --- Резолюция risk-guard (заполняется при review-risk) ---
        "risk_guard": {
            "status": "pending",       # pending | approved | rejected
            "reviewed_at": None,
            "verdict": None,           # отклонено | открываем
            "reason": None,
            "checks": [],
        },

        # --- Параметры закрытия (заполняются при close) ---
        "close": {
            "time": None,
            "price": None,
            "reason": None,
            "exit_decision": None,
        },

        # --- P&L (заполняется при close) ---
        "pnl": {
            "pnl_rub": None,
            "pnl_pct": None,
            "entry_commission": 0.0,
            "close_commission": 0.0,
            "total_commission": 0.0,
            "net_pnl_rub": None,
            "net_pnl_pct": None,
            "roi": None,
        },

        # --- Метаданные ---
        "metadata": {
            "record_created": _now_str(),
            "record_updated": _now_str(),
            "record_file": filename,
            "status": "pending",  # pending | open | closed | rejected
        },
    }


def save_trade_record(ticker: str, direction: str) -> str:
    """Создаёт новый артефакт сделки с пустой структурой.

    Args:
        ticker: Тикер инструмента.
        direction: "LONG" или "SHORT".

    Returns:
        Путь к сохранённому файлу.
    """
    record = _new_trade_artifact(ticker, direction)
    filepath = TRADES_DIR / record["metadata"]["record_file"]
    filepath.write_text(json.dumps(record, indent=2, ensure_ascii=False), encoding="utf-8")
    return str(filepath)


def _load_trade_file(ticker: str, trade_date: Optional[str] = None) -> Optional[dict]:
    """Загружает артефакт сделки по тикеру и дате.

    Args:
        ticker: Тикер.
        trade_date: Дата сделки (YYYY-MM-DD). По умолчанию сегодня.

    Returns:
        Словарь с данными сделки или None.
    """
    if trade_date is None:
        trade_date = _today_str()
    filename = f"trade_{trade_date}_{ticker.upper()}.json"
    filepath = TRADES_DIR / filename
    if filepath.exists():
        try:
            return json.loads(filepath.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, Exception):
            return None
    return None


def _save_trade_artifact(record: dict) -> str:
    """Сохраняет/обновляет артефакт сделки на диск.

    Args:
        record: Словарь с данными сделки.

    Returns:
        Путь к сохранённому файлу.
    """
    record["metadata"]["record_updated"] = _now_str()
    filepath = TRADES_DIR / record["metadata"]["record_file"]
    filepath.write_text(json.dumps(record, indent=2, ensure_ascii=False), encoding="utf-8")
    return str(filepath)


def update_trade_entry(
    ticker: str,
    entry_price: float,
    volume: int,
    deposit_at_entry: float,
    levels: Optional[dict] = None,
    position_sizing: Optional[dict] = None,
    rationale: Optional[str] = None,
    signal_type: Optional[str] = None,
    signal_strength: Optional[int] = None,
    total_votes: Optional[int] = None,
    morning_verdict: Optional[str] = None,
    trade_date: Optional[str] = None,
) -> Optional[str]:
    """Записывает параметры входа в артефакт сделки.

    Args:
        ticker: Тикер.
        entry_price: Цена входа.
        volume: Количество акций/лотов.
        deposit_at_entry: Размер депозита на момент входа.
        levels: Словарь с уровнями SL/TP (опционально).
        position_sizing: Словарь с расчётом размера позиции (опционально).
        rationale: Обоснование входа.
        signal_type: Тип сигнала (rsi, macd, volume и т.д.).
        signal_strength: Сила сигнала.
        total_votes: Количество голосов комбинированного сигнала.
        morning_verdict: Вердикт утреннего подтверждения.
        trade_date: Дата сделки. По умолчанию сегодня.

    Returns:
        Путь к файлу или None, если артефакт не найден.
    """
    record = _load_trade_file(ticker, trade_date)
    if record is None:
        return None

    entry_value = entry_price * volume

    record["entry"]["time"] = _now_str()
    record["entry"]["price"] = entry_price
    record["entry"]["volume"] = volume
    record["entry"]["value"] = entry_value
    record["entry"]["deposit_at_entry"] = deposit_at_entry
    record["entry"]["rationale"] = rationale
    record["entry"]["signal_type"] = signal_type
    record["entry"]["signal_strength"] = signal_strength
    record["entry"]["total_votes"] = total_votes
    record["entry"]["morning_verdict"] = morning_verdict

    if levels:
        record["levels"]["stop_loss"] = levels.get("stop_loss")
        record["levels"]["take_profit"] = levels.get("take_profit")
        record["levels"]["risk_pct"] = levels.get("risk_pct")
        record["levels"]["reward_pct"] = levels.get("reward_pct")
        record["levels"]["risk_reward_ratio"] = levels.get("risk_reward_ratio")

    if position_sizing:
        record["position_sizing"]["risk_per_trade_pct"] = position_sizing.get("risk_per_trade_pct")
        record["position_sizing"]["max_position_pct"] = position_sizing.get("max_position_pct")
        record["position_sizing"]["position_pct"] = position_sizing.get("position_pct")
        record["position_sizing"]["risk_amount_rub"] = position_sizing.get("risk_amount_rub")
        record["position_sizing"]["recommended_volume"] = position_sizing.get("recommended_volume", volume)

    record["metadata"]["status"] = "open"

    return _save_trade_artifact(record)


def update_trade_risk_guard(
    ticker: str,
    verdict: str,
    reason: Optional[str] = None,
    checks: Optional[list] = None,
    trade_date: Optional[str] = None,
) -> Optional[str]:
    """Записывает резолюцию risk-guard в артефакт сделки.

    Args:
        ticker: Тикер.
        verdict: Резолюция: "отклонено" или "открываем".
        reason: Причина решения.
        checks: Список пройденных/проваленных проверок.
        trade_date: Дата сделки. По умолчанию сегодня.

    Returns:
        Путь к файлу или None, если артефакт не найден.
    """
    record = _load_trade_file(ticker, trade_date)
    if record is None:
        return None

    record["risk_guard"]["status"] = "approved" if verdict == "открываем" else "rejected"
    record["risk_guard"]["reviewed_at"] = _now_str()
    record["risk_guard"]["verdict"] = verdict
    record["risk_guard"]["reason"] = reason
    record["risk_guard"]["checks"] = checks or []

    if verdict == "отклонено":
        record["metadata"]["status"] = "rejected"

    return _save_trade_artifact(record)


def update_trade_close(
    ticker: str,
    close_price: float,
    exit_reason: str = "closed_manual",
    exit_decision: Optional[str] = None,
    commission_rate: float = 0.0005,
    trade_date: Optional[str] = None,
) -> Optional[str]:
    """Закрывает сделку с расчётом P&L.

    Args:
        ticker: Тикер.
        close_price: Цена закрытия.
        exit_reason: Причина закрытия (closed_by_sl, closed_by_tp, closed_manual).
        exit_decision: Решение о закрытии (держать/закрываем из утреннего анализа).
        commission_rate: Ставка комиссии (0.05% по умолчанию).
        trade_date: Дата сделки. По умолчанию сегодня.

    Returns:
        Путь к файлу или None, если артефакт не найден.
    """
    record = _load_trade_file(ticker, trade_date)
    if record is None:
        return None

    entry_price = record["entry"]["price"]
    volume = record["entry"]["volume"]
    direction = record["direction"]

    if entry_price is None or entry_price <= 0:
        return None

    # P&L в процентах
    pnl_pct = ((close_price - entry_price) / entry_price) * 100
    if direction == "SHORT":
        pnl_pct = -pnl_pct

    # P&L в рублях
    entry_value = entry_price * volume
    close_value = close_price * volume
    if direction == "LONG":
        pnl_rub = close_value - entry_value
    else:
        pnl_rub = entry_value - close_value

    # Комиссии
    entry_commission = entry_value * commission_rate
    close_commission = close_value * commission_rate
    total_commission = entry_commission + close_commission

    # Чистая прибыль
    net_pnl_rub = pnl_rub - total_commission
    net_pnl_pct = (net_pnl_rub / entry_value) * 100 if entry_value > 0 else 0

    # ROI (относительно депозита на момент входа)
    deposit_at_entry = record["entry"]["deposit_at_entry"]
    roi = (net_pnl_rub / deposit_at_entry * 100) if deposit_at_entry and deposit_at_entry > 0 else None

    record["close"]["time"] = _now_str()
    record["close"]["price"] = close_price
    record["close"]["reason"] = exit_reason
    record["close"]["exit_decision"] = exit_decision

    record["pnl"]["pnl_rub"] = round(pnl_rub, 2)
    record["pnl"]["pnl_pct"] = round(pnl_pct, 2)
    record["pnl"]["entry_commission"] = round(entry_commission, 2)
    record["pnl"]["close_commission"] = round(close_commission, 2)
    record["pnl"]["total_commission"] = round(total_commission, 2)
    record["pnl"]["net_pnl_rub"] = round(net_pnl_rub, 2)
    record["pnl"]["net_pnl_pct"] = round(net_pnl_pct, 2)
    record["pnl"]["roi"] = round(roi, 2) if roi is not None else None

    record["metadata"]["status"] = exit_reason

    return _save_trade_artifact(record)


# ---------------------------------------------------------------------------
# Совместимость со старым кодом (заглушки)
# ---------------------------------------------------------------------------

def close_trade(ticker: str, close_price: float, close_time: Optional[str] = None, status: str = "closed_manual") -> bool:
    """Закрывает открытую сделку (обёртка над update_trade_close для совместимости).

    Args:
        ticker: Тикер.
        close_price: Цена закрытия.
        close_time: Время закрытия (не используется, берётся текущее).
        status: Причина закрытия.

    Returns:
        True если сделка найдена и закрыта.
    """
    result = update_trade_close(ticker, close_price, exit_reason=status)
    return result is not None


# ---------------------------------------------------------------------------
# Загрузка артефактов
# ---------------------------------------------------------------------------

def load_evening_scan(scan_date: Optional[str] = None) -> Optional[dict]:
    """Загружает вечерний артефакт (JSON) за указанную дату.

    Args:
        scan_date: Дата в формате YYYY-MM-DD. По умолчанию сегодня.

    Returns:
        Словарь с данными артефакта или None
    """
    if scan_date is None:
        scan_date = _today_str()

    filepath = ARTIFACTS_DIR / f"evening_scan_{scan_date}.json"
    if filepath.exists():
        try:
            return json.loads(filepath.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, Exception):
            return None
    return None


def load_morning_confirmation(confirm_date: Optional[str] = None) -> Optional[dict]:
    """Загружает утреннее подтверждение (JSON) за указанную дату."""
    if confirm_date is None:
        confirm_date = _today_str()

    filepath = ARTIFACTS_DIR / f"morning_confirmation_{confirm_date}.json"
    if filepath.exists():
        try:
            return json.loads(filepath.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, Exception):
            return None
    return None


def load_latest_artifact(artifact_type: str) -> Optional[dict]:
    """Загружает последний артефакт указанного типа (JSON).

    Args:
        artifact_type: "evening_scan" или "morning_confirmation"

    Returns:
        Словарь с данными последнего артефакта или None
    """
    pattern = f"{artifact_type}_*.json"
    files = sorted(ARTIFACTS_DIR.glob(pattern), reverse=True)
    if files:
        try:
            return json.loads(files[0].read_text(encoding="utf-8"))
        except (json.JSONDecodeError, Exception):
            return None
    return None


def load_trades(from_date: Optional[str] = None, to_date: Optional[str] = None) -> list:
    """Загружает историю сделок.

    Args:
        from_date: Начальная дата (YYYY-MM-DD). По умолчанию все.
        to_date: Конечная дата (YYYY-MM-DD). По умолчанию все.

    Returns:
        Список словарей с данными сделок
    """
    trades = []
    for filepath in sorted(TRADES_DIR.glob("trade_*.json")):
        try:
            data = json.loads(filepath.read_text(encoding="utf-8"))
            trades.append(data)
        except (json.JSONDecodeError, Exception):
            continue

    # Фильтрация по дате (поддерживаем разные форматы полей)
    if from_date or to_date:
        filtered = []
        for t in trades:
            # Пробуем разные варианты поля с датой
            trade_date = (
                t.get("entry_date", "")
                or t.get("entry_time", "")
                or t.get("record_created", "")
            )
            trade_date = trade_date[:10]  # обрезаем до YYYY-MM-DD
            if from_date and trade_date < from_date:
                continue
            if to_date and trade_date > to_date:
                continue
            filtered.append(t)
        trades = filtered

    return trades


def load_open_trades() -> list:
    """Загружает все открытые сделки.

    Ищет сделки со статусом "open" в metadata.status.
    """
    return [t for t in load_trades() if t.get("metadata", {}).get("status") == "open"]


def load_pending_trades() -> list:
    """Загружает сделки, ожидающие валидации risk-guard.

    Returns:
        Список сделок с risk_guard.status == "pending".
    """
    return [
        t for t in load_trades()
        if t.get("risk_guard", {}).get("status") == "pending"
        and t.get("metadata", {}).get("status") == "pending"
    ]


# (вспомогательные функции для генерации markdown удалены,
#  так как .md теперь формирует AI-агент из JSON)