197 lines
6.6 KiB
Python
197 lines
6.6 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
|
|
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
|
|
icon: str = ""
|
|
alert_early_points: float | None = None
|
|
|
|
@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 in {"comment", "symbol"}
|
|
|
|
@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}"
|
|
direction = "up" if self.side is Side.SUPPORT else "down"
|
|
return f"{direction}{self.tf.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,
|
|
alert_early_points=self.alert_early_points,
|
|
)
|
|
|
|
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)),
|
|
icon=str(value.get("icon", "")),
|
|
alert_early_points=(
|
|
float(value["alert_early_points"])
|
|
if value.get("alert_early_points") 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]:
|
|
"""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)
|