from __future__ import annotations
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from .constants import TREND_BEAR, TREND_BULL, TREND_NEUTRAL
from .time_display import format_timestamp
def _is_intraday_interval(interval: str) -> bool:
return interval in {"1m", "2m", "5m", "15m", "30m", "60m", "90m", "1h"}
def _is_daily_interval(interval: str) -> bool:
return interval == "1d"
def _intraday_day_ticks(index: pd.DatetimeIndex) -> tuple[list[pd.Timestamp], list[str]]:
if len(index) == 0:
return [], []
normalized = index.normalize()
first_mask = ~normalized.duplicated()
tickvals = [index[pos] for pos, keep in enumerate(first_mask) if bool(keep)]
ticktext = [f"{ts.month}/{ts.day}" for ts in tickvals]
return tickvals, ticktext
def _missing_calendar_day_values(df: pd.DataFrame) -> list[str]:
if df.empty:
return []
index = pd.DatetimeIndex(df.index)
session_days = pd.DatetimeIndex(index.normalize().unique()).sort_values()
if len(session_days) < 2:
return []
if session_days.tz is None:
all_days = pd.date_range(start=session_days[0], end=session_days[-1], freq="D")
else:
all_days = pd.date_range(start=session_days[0], end=session_days[-1], freq="D", tz=session_days.tz)
missing_days = all_days.difference(session_days)
# Weekend gaps are already handled by sat->mon bounds; keep explicit values
# for weekday closures (e.g., exchange holidays) to avoid overlap artifacts.
weekday_missing = [day for day in missing_days if day.dayofweek < 5]
return [day.strftime("%Y-%m-%d") for day in weekday_missing]
def build_figure(
df: pd.DataFrame,
gray_fake: bool,
*,
interval: str,
hide_market_closed_gaps: bool,
display_timezone: str,
use_24h_time: bool,
) -> go.Figure:
fig = make_subplots(
rows=2,
cols=1,
row_heights=[0.8, 0.2],
vertical_spacing=0.03,
shared_xaxes=True,
)
bull_mask = df["classification"] == "real_bull"
bear_mask = df["classification"] == "real_bear"
time_labels = [
format_timestamp(ts, display_timezone=display_timezone, use_24h_time=use_24h_time) for ts in df.index
]
if gray_fake:
fig.add_trace(
go.Candlestick(
x=df.index,
open=df["Open"],
high=df["High"],
low=df["Low"],
close=df["Close"],
name="All Bars",
increasing_line_color="#B0B0B0",
decreasing_line_color="#808080",
opacity=0.35,
customdata=time_labels,
hovertemplate=(
"Time: %{customdata}
"
"Open: %{open:.2f}
"
"High: %{high:.2f}
"
"Low: %{low:.2f}
"
"Close: %{close:.2f}All Bars"
),
),
row=1,
col=1,
)
else:
fig.add_trace(
go.Candlestick(
x=df.index,
open=df["Open"],
high=df["High"],
low=df["Low"],
close=df["Close"],
name="All Bars",
increasing_line_color="#2E8B57",
decreasing_line_color="#B22222",
opacity=0.6,
customdata=time_labels,
hovertemplate=(
"Time: %{customdata}
"
"Open: %{open:.2f}
"
"High: %{high:.2f}
"
"Low: %{low:.2f}
"
"Close: %{close:.2f}All Bars"
),
),
row=1,
col=1,
)
bull_time_labels = [time_labels[idx] for idx, is_bull in enumerate(bull_mask) if bool(is_bull)]
fig.add_trace(
go.Scatter(
x=df.index[bull_mask],
y=df.loc[bull_mask, "Close"],
mode="markers",
name="Real Bullish",
marker=dict(color="#00C853", size=9, symbol="triangle-up"),
customdata=bull_time_labels,
hovertemplate="Time: %{customdata}
Close: %{y:.2f}Real Bullish",
),
row=1,
col=1,
)
bear_time_labels = [time_labels[idx] for idx, is_bear in enumerate(bear_mask) if bool(is_bear)]
fig.add_trace(
go.Scatter(
x=df.index[bear_mask],
y=df.loc[bear_mask, "Close"],
mode="markers",
name="Real Bearish",
marker=dict(color="#D50000", size=9, symbol="triangle-down"),
customdata=bear_time_labels,
hovertemplate="Time: %{customdata}
Close: %{y:.2f}Real Bearish",
),
row=1,
col=1,
)
trend_color = df["trend_state"].map(
{
TREND_BULL: "#00C853",
TREND_BEAR: "#D50000",
TREND_NEUTRAL: "#9E9E9E",
}
)
fig.add_trace(
go.Bar(
x=df.index,
y=df["Volume"],
marker_color=trend_color,
name="Volume",
opacity=0.65,
customdata=time_labels,
hovertemplate="Time: %{customdata}
Volume: %{y}Volume",
),
row=2,
col=1,
)
fig.update_layout(
template="plotly_white",
xaxis_rangeslider_visible=False,
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="left", x=0),
margin=dict(l=20, r=20, t=60, b=20),
height=760,
)
if hide_market_closed_gaps:
if _is_intraday_interval(interval):
# Intraday rangebreak combinations can produce axis rendering artifacts
# with some feeds/timezones. Categorical axis keeps chronological bars
# contiguous and removes closed-session gaps reliably.
tickvals, ticktext = _intraday_day_ticks(pd.DatetimeIndex(df.index))
fig.update_xaxes(
type="category",
categoryorder="array",
categoryarray=list(df.index),
tickmode="array",
tickvals=tickvals,
ticktext=ticktext,
tickangle=0,
)
elif _is_daily_interval(interval):
rangebreaks: list[dict[str, object]] = [dict(bounds=["sat", "mon"])]
missing_days = _missing_calendar_day_values(df)
if missing_days:
rangebreaks.append(dict(values=missing_days))
fig.update_xaxes(rangebreaks=rangebreaks)
fig.update_yaxes(title_text="Price", row=1, col=1)
fig.update_yaxes(title_text="Volume", row=2, col=1)
return fig