chart/app/analysis/manual_lines.py

import json
from dataclasses import asdict, dataclass, replace
from pathlib import Path
from threading import RLock

from app.analysis.levels import Level, LevelKind, Side
from app.bars.models import Timeframe
from app.config import TIMEFRAME_WEIGHT


@dataclass(slots=True)
class ManualLine:
    id: str
    tf: Timeframe
    side: Side
    anchor_t: int
    anchor_p: float
    slope: float
    last_t: int
    created_at: int
    note: str = ""
    hidden: bool = False
    color: str = "#65b7cf"
    line_width: int = 2
    number: int = 0
    cutoff_t: int | None = None
    armed: bool = True

    @property
    def horizontal(self) -> bool:
        """A typed price level rather than a drawn trendline."""
        return self.slope == 0.0

    def default_label(self) -> str:
        if self.horizontal:
            return f"@ {self.anchor_p:.2f}"
        return f"{self.tf.value} {self.side.value}"

    def to_level(self) -> Level:
        return Level(
            id=self.id,
            kind=LevelKind.MANUAL,
            tf=self.tf,
            side=self.side,
            weight=TIMEFRAME_WEIGHT[self.tf],
            score=1.0,
            label=self.note or self.default_label(),
            anchor_t=self.anchor_t,
            anchor_p=self.anchor_p,
            slope=self.slope,
            points=None,
            touches=0,
            first_t=self.anchor_t,
            last_t=self.last_t,
            provisional=False,
            hidden=self.hidden,
            color=self.color,
            line_width=self.line_width,
            number=self.number,
            cutoff_t=self.cutoff_t,
            armed=self.armed,
        )

    def to_dict(self) -> dict:
        value = asdict(self)
        value["tf"] = self.tf.value
        value["side"] = self.side.value
        return value

    @classmethod
    def from_dict(cls, value: dict) -> "ManualLine":
        return cls(
            id=str(value["id"]),
            tf=Timeframe(value["tf"]),
            side=Side(value["side"]),
            anchor_t=int(value["anchor_t"]),
            anchor_p=float(value["anchor_p"]),
            slope=float(value["slope"]),
            last_t=int(value.get("last_t", value["anchor_t"])),
            created_at=int(value["created_at"]),
            note=str(value.get("note", "")),
            hidden=bool(value.get("hidden", False)),
            color=str(value.get("color", "#65b7cf")),
            line_width=int(value.get("line_width", 2)),
            number=int(value.get("number", 0)),
            cutoff_t=int(value["cutoff_t"]) if value.get("cutoff_t") is not None else None,
            armed=bool(value.get("armed", True)),
        )


class ManualLineStore:
    def __init__(self, path: str | Path):
        self.path = Path(path)
        self._lock = RLock()
        loaded = self.load()
        next_number = max((line.number for line in loaded), default=0)
        for index, line in enumerate(loaded):
            if line.number <= 0:
                next_number += 1
                loaded[index] = replace(line, number=next_number)
        self.lines: dict[str, ManualLine] = {line.id: line for line in loaded}

    def load(self) -> list[ManualLine]:
        if not self.path.exists():
            return []
        payload = json.loads(self.path.read_text(encoding="utf-8"))
        return [ManualLine.from_dict(value) for value in payload]

    def save(self) -> None:
        with self._lock:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            temporary = self.path.with_suffix(self.path.suffix + ".tmp")
            temporary.write_text(
                json.dumps(
                    [line.to_dict() for line in self.lines.values()],
                    indent=2,
                    sort_keys=True,
                )
                + "\n",
                encoding="utf-8",
            )
            temporary.replace(self.path)

    def add(self, line: ManualLine) -> ManualLine:
        with self._lock:
            if line.number <= 0:
                line = replace(line, number=max((value.number for value in self.lines.values()), default=0) + 1)
            self.lines[line.id] = line
            self.save()
            return line

    def update(self, line_id: str, changes: dict) -> ManualLine:
        with self._lock:
            if line_id not in self.lines:
                raise KeyError(line_id)
            line = replace(self.lines[line_id], **changes)
            self.lines[line_id] = line
            self.save()
            return line

    def delete(self, line_id: str) -> None:
        with self._lock:
            if line_id not in self.lines:
                raise KeyError(line_id)
            del self.lines[line_id]
            self.save()

    def levels(self) -> list[Level]:
        return [line.to_level() for line in self.lines.values()]