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
    # "" means a line predating drawing kinds; `kind` derives it. Comments carry
    # their text in `note`, like every other drawing carries its label.
    kind: str = ""
    pinned: bool = True
    x: float = 0.72
    y: float = 0.12
    collapsed: bool = False

    @property
    def drawing_kind(self) -> str:
        """What this drawing is, for grouping and filtering.

        Derived when absent so drawings saved before comments existed keep
        working: a zero slope was always a typed price level, anything else a
        drawn trendline.
        """
        if self.kind:
            return self.kind
        return "level" if self.horizontal else "trendline"

    @property
    def is_comment(self) -> bool:
        return self.drawing_kind == "comment"

    @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)),
            kind=str(value.get("kind", "")),
            pinned=bool(value.get("pinned", True)),
            x=float(value.get("x", 0.72)),
            y=float(value.get("y", 0.12)),
            collapsed=bool(value.get("collapsed", False)),
        )


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]:
        """Only the drawings that are levels.

        Comments are stored alongside lines so they share numbering, filtering
        and deletion, but they are annotations. Letting one through here would
        put it in a confluence cluster and fire a push notification about a
        piece of text.
        """
        return [line.to_level() for line in self.lines.values() if not line.is_comment]

    def drawings(self) -> list[ManualLine]:
        """Everything drawn, comments included, newest number last."""
        return sorted(self.lines.values(), key=lambda line: line.number)