136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
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
|
|
|
|
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 f"{self.tf.value} {self.side.value}",
|
|
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,
|
|
)
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
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()]
|