62 lines
2.3 KiB
Python
62 lines
2.3 KiB
Python
from app.analysis.indicators import ema, sma
|
|
from app.analysis.levels import Level, LevelKind, Side
|
|
from app.bars.models import Bar, Timeframe
|
|
from app.config import MA_WEIGHT_FACTOR, TIMEFRAME_WEIGHT
|
|
|
|
|
|
MA_FUNCTIONS = {"sma": sma, "ema": ema}
|
|
|
|
|
|
def build_ma_levels(
|
|
bars_by_tf: dict[Timeframe, list[Bar]],
|
|
ma_sets: dict[Timeframe, list[tuple[str, int]]],
|
|
) -> list[Level]:
|
|
levels: list[Level] = []
|
|
for tf, definitions in ma_sets.items():
|
|
bars = bars_by_tf.get(tf, [])
|
|
closed_count = sum(bar.closed for bar in bars)
|
|
closes = [bar.c for bar in bars]
|
|
for kind, period in definitions:
|
|
if closed_count < period or kind not in MA_FUNCTIONS:
|
|
continue
|
|
values = MA_FUNCTIONS[kind](closes, period)
|
|
points = [(bar.t, value) for bar, value in zip(bars, values) if value is not None]
|
|
if not points:
|
|
continue
|
|
current = points[-1][1]
|
|
provisional = not bars[-1].closed
|
|
levels.append(
|
|
Level(
|
|
id=f"ma:{tf.value}:{kind}:{period}",
|
|
kind=LevelKind.MA,
|
|
tf=tf,
|
|
side=Side.SUPPORT if current <= bars[-1].c else Side.RESISTANCE,
|
|
weight=TIMEFRAME_WEIGHT[tf] * MA_WEIGHT_FACTOR,
|
|
score=1.0,
|
|
label=f"{tf.value} {kind.upper()}{period}",
|
|
anchor_t=points[-1][0],
|
|
anchor_p=current,
|
|
slope=0.0,
|
|
points=points,
|
|
touches=0,
|
|
first_t=points[0][0],
|
|
last_t=points[-1][0],
|
|
provisional=provisional,
|
|
hidden=False,
|
|
period=period,
|
|
)
|
|
)
|
|
return levels
|
|
|
|
|
|
def project_step(points: list[tuple[int, float]], bars: list[Bar]) -> list[tuple[int, float]]:
|
|
projected: list[tuple[int, float]] = []
|
|
point_index = 0
|
|
current: float | None = None
|
|
for bar in bars:
|
|
while point_index < len(points) and points[point_index][0] <= bar.t:
|
|
current = points[point_index][1]
|
|
point_index += 1
|
|
if current is not None:
|
|
projected.append((bar.t, current))
|
|
return projected
|