diff --git a/src/assets/clientside.js b/src/assets/clientside.js
index 9df9f28..88027bb 100644
--- a/src/assets/clientside.js
+++ b/src/assets/clientside.js
@@ -319,6 +319,13 @@ window.dash_clientside = Object.assign({}, window.dash_clientside, {
}
var liveAxes = {};
(figure.data || []).forEach(function(t) { liveAxes[axisKeyOf(t)] = true; });
+ // A heatmap's axis is a grid, not a scale: its rows are categories or fixed
+ // lanes (the /categories flow panel and state strip), and fitting it to
+ // the points in a window paired lane numbers with report dates and turned
+ // the strip upside down. Leave every axis that carries one alone.
+ (figure.data || []).forEach(function(t) {
+ if (t.type === 'heatmap') { delete liveAxes[axisKeyOf(t)]; }
+ });
// A figure may ship its own rules in layout.meta.refit: which axes are
// fitted at all, and the ratio at which a price panel earns a log scale.
diff --git a/src/components/category_traces.py b/src/components/category_traces.py
index 88bf068..42b4011 100644
--- a/src/components/category_traces.py
+++ b/src/components/category_traces.py
@@ -438,8 +438,9 @@ def _composite_over(rgba, hex_bg):
]
# One heatmap row: the tick label the axis shows, the fuller name the hover opens
-# with, and the columns the cells and hover read.
-FlowRow = namedtuple("FlowRow", "label hover_label z dnet dlong dshort thin")
+# with, and the columns the cells, the hover and the level markers read.
+FlowRow = namedtuple("FlowRow",
+ "label hover_label z dnet dlong dshort thin level mark")
def _counterparty_label(df):
@@ -472,7 +473,7 @@ def _counterparty_label(df):
return f"Counterparty ({' + '.join(names)})"
-def _flow_rows(df, series, with_counterparty=True):
+def _flow_rows(df, series, lookback_header=None, with_counterparty=True):
"""The rows to draw, in order: the selected cohorts, then the counterparty.
The composite row is drawn whether or not the checklist selected its members.
@@ -487,6 +488,10 @@ def _flow_rows(df, series, with_counterparty=True):
took 243 px on gold and 334 px on silver, a quarter to a third of the width from
every panel, not only this one. The hover has no such cost, so that is where the
members are named.
+
+ The level columns carry the page's lookback header (cotmetrics.flows: the level
+ is the range index the page draws, one week earlier), so without a header a row
+ simply has no markers.
"""
rows = []
for s in series:
@@ -501,6 +506,10 @@ def _flow_rows(df, series, with_counterparty=True):
dlong=flows.flow_long_col(s.spec),
dshort=flows.flow_short_col(s.spec),
thin=flows.flow_thin_col(s.spec),
+ level=(flows.flow_from_level_col(s.spec, lookback_header)
+ if lookback_header is not None else None),
+ mark=(flows.flow_level_mark_col(s.spec, lookback_header)
+ if lookback_header is not None else None),
))
z = flows.counterparty_flow_z_col()
if with_counterparty and z in df.columns:
@@ -509,9 +518,9 @@ def _flow_rows(df, series, with_counterparty=True):
hover_label=_counterparty_label(df),
z=z,
dnet=flows.counterparty_flow_col(),
- # The composite is a sum of nets, so it has no legs and no thin flag of
- # its own; the hover for it stops at the net and the z.
- dlong=None, dshort=None, thin=None,
+ # The composite is a sum of nets, so it has no legs, no thin flag and
+ # no range index of its own; the hover for it stops at the net and the z.
+ dlong=None, dshort=None, thin=None, level=None, mark=None,
))
return rows
@@ -538,10 +547,16 @@ def _flow_hover_text(df, rows):
board's ranking is of index-point changes in the Legacy Commercial positioning
index, a different quantity on a different report, and the two must not read as
the same thing.
+
+ A marked cell (see _level_markers) adds the level its flow departed from, and
+ nothing about what came after it.
"""
def _bool(v):
return (not pd.isna(v)) and bool(v)
+ weeks = df.attrs.get("flow_level_weeks") or df.attrs.get("lookback_weeks")
+ span = f"{weeks}-week range" if weeks else "lookback range"
+
out = []
for row in rows:
z = df[row.z]
@@ -549,6 +564,8 @@ def _bool(v):
dlong = df[row.dlong] if row.dlong and row.dlong in df.columns else None
dshort = df[row.dshort] if row.dshort and row.dshort in df.columns else None
thin = df[row.thin] if row.thin and row.thin in df.columns else None
+ mark = df[row.mark] if row.mark and row.mark in df.columns else None
+ level = df[row.level] if row.level and row.level in df.columns else None
cells = []
for i, date in enumerate(df.index):
zv = z.iloc[i]
@@ -571,18 +588,69 @@ def _bool(v):
if thin is not None and _bool(thin.iloc[i]):
text += (f"
thin: typical week under "
f"{const.FLOW_MIN_STD_CONTRACTS} contracts, read the count")
+ if mark is not None and level is not None and _bool(mark.iloc[i]) \
+ and not pd.isna(level.iloc[i]):
+ text += f"
from level {level.iloc[i]:.1f} of the {span}"
cells.append(text)
out.append(cells)
return out
-def _flow_heatmap(fig, df, rows, row, col, facet=False):
+# The marker sits on the cell, so it has to read on cyan, orange and grey alike:
+# near-white with a background-coloured edge. Never a palette slot, for the reason
+# FLOW_COLORSCALE gives.
+FLOW_MARK_COLOR = vc.BRIGHTER_TEXT_COLOR
+FLOW_MARK_EDGE = vc.BACKGROUND_COLOR
+
+
+def _level_markers(fig, df, rows, row, col, facet=False):
+ """A triangle on each cell whose active flow left an extreme of the range.
+
+ Up for the top of the range (the prior week's index above
+ const.FLOW_LEVEL_HIGH), down for the bottom (below FLOW_LEVEL_LOW); the cell's
+ colour already says which way the cohort moved. The marks come from cotmetrics
+ (`flows.flow_level_mark_col`), which read the page's own range index, so the
+ level a marker points at is the level the Positioning Index panel draws.
+
+ Hover is skipped: the cell under the marker already carries the level in its
+ hover, and under hovermode "x unified" a second entry would repeat it.
+ Context for reading a cell; nothing here says what followed a flow from an
+ extreme, because the cells that motivated the cutoffs were descriptive and on
+ gold alone.
+ """
+ xs, ys, symbols = [], [], []
+ for r in rows:
+ if not r.mark or r.mark not in df.columns:
+ continue
+ mark = df[r.mark]
+ hit = mark.notna().to_numpy() & (mark.fillna(0).to_numpy() != 0)
+ for date, m in zip(df.index[hit], mark[hit]):
+ xs.append(date)
+ ys.append(r.label)
+ symbols.append("triangle-up" if m > 0 else "triangle-down")
+ if not xs:
+ return fig
+ fig.add_trace(go.Scatter(
+ x=xs, y=ys, mode="markers",
+ marker=dict(symbol=symbols, size=6 if facet else 7, color=FLOW_MARK_COLOR,
+ line=dict(width=1, color=FLOW_MARK_EDGE)),
+ hoverinfo="skip",
+ name="Level marks",
+ showlegend=False,
+ ), row=row, col=col)
+ return fig
+
+
+def _flow_heatmap(fig, df, rows, row, col, facet=False, dense=None):
# The gap is a property of the cell width, not of the trace. Measured at an
# 857 px figure: the overlay gives 3.7 px a week over the 156-week window, so a
# 1 px gap still leaves cells; a facet column gives 2.1 px a week (328 px for
# the column), where 1 px is half of every cell and the panel reads as
# hairlines; a phone opens on 52 weeks at about 6 px a cell, a sixth of it gap.
- dense = facet or app_utils.is_mobile()
+ # The flow view passes dense=True: in the running app its full-width panel at
+ # one pixel of gap a week read as a comb rather than as runs of colour.
+ if dense is None:
+ dense = facet or app_utils.is_mobile()
fig.add_trace(go.Heatmap(
x=df.index,
y=[r.label for r in rows],
@@ -611,7 +679,7 @@ def _flow_heatmap(fig, df, rows, row, col, facet=False):
autorange="reversed", fixedrange=True, showgrid=False,
zeroline=False, tickfont=dict(size=9), title=None,
showticklabels=not facet)
- return fig
+ return _level_markers(fig, df, rows, row, col, facet=facet)
def get_category_flow_heatmap(fig, df, series, lookback_header, row, col, palette,
@@ -634,7 +702,7 @@ def get_category_flow_heatmap(fig, df, series, lookback_header, row, col, palett
Colour is polarity, never a palette slot, and the state and z are caption
vocabulary: nothing here carries a forward return or a verdict.
"""
- rows = _flow_rows(df, series)
+ rows = _flow_rows(df, series, lookback_header)
if not rows:
return fig
_flow_heatmap(fig, df, rows, row, col)
@@ -646,15 +714,196 @@ def get_category_flow_row(fig, df, series, lookback_header, row, col, palette,
"""The facet form: a one-row heatmap for the single cohort in the cell.
No composite row here. In small multiples every row is one category and the
- counterparty is context, which is a row of its own (the next PR's context row),
- the same way price and open interest are rows rather than overlays.
+ counterparty is context, a row of its own under price (facet_context_rows), the
+ same way price and open interest are rows rather than overlays.
+ """
+ rows = _flow_rows(df, series, lookback_header, with_counterparty=False)
+ if not rows:
+ return fig
+ return _flow_heatmap(fig, df, rows, row, col, facet=True)
+
+
+def get_counterparty_flow_row(fig, df, row, col):
+ """The facet context row: the counterparty composite as a one-row heatmap.
+
+ The overlay panel ends with this row; in small multiples it is its own row
+ under the category rows, drawn only in the flow column, because the composite
+ has a flow and no positioning series for the other panels to draw.
"""
- rows = _flow_rows(df, series, with_counterparty=False)
+ rows = _flow_rows(df, [])
if not rows:
return fig
return _flow_heatmap(fig, df, rows, row, col, facet=True)
+# The strip paints signs, not z: three values, so three exact colours from the same
+# polarity pair as the heatmap, with the grey midpoint for "inside one sd".
+FLOW_SIGN_COLORSCALE = FLOW_COLORSCALE
+FLOW_SPLIT_COLOR = vc.BRIGHTER_TEXT_COLOR
+# The split-week ticks ride a fourth lane above the three, at this y.
+_SPLIT_LANE_Y = -0.9
+
+
+def _opinion_specs(df):
+ """The three cohorts the state is read from, in the order cotmetrics used.
+
+ From the frame's own roles, not the checklist: the state is a fact about all
+ three whether or not the reader switched one off.
+ """
+ roles = df.attrs.get("flow_roles") or {}
+ report = roles.get("report")
+ if report not in categories.REPORT_CHOICES:
+ return []
+ by_key = {s.key: s for s in categories.categories_for(report)}
+ specs = [by_key[k] for k in roles.get("opinion") or () if k in by_key]
+ return [s for s in specs if flows.flow_sign_col(s) in df.columns]
+
+
+def has_state_strip(df):
+ """True when the frame carries a state to draw: a state-eligible market."""
+ return (df is not None and const.FLOW_STATE in df.columns
+ and len(_opinion_specs(df)) == 3)
+
+
+def _strip_hover_text(df, specs):
+ states = df[const.FLOW_STATE]
+ out = []
+ for s in specs:
+ z = df[flows.flow_z_col(s)]
+ sign = df[flows.flow_sign_col(s)]
+ cells = []
+ for i in range(len(df)):
+ if pd.isna(sign.iloc[i]):
+ cells.append(None)
+ continue
+ v = int(sign.iloc[i])
+ zv = z.iloc[i]
+ move = ("net buying" if v > 0 else "net selling" if v < 0
+ else "inside one sd")
+ state = states.iloc[i]
+ head = (f"{state}: a vocabulary label, not a signal"
+ if isinstance(state, str) else "no state this week")
+ # No date line: the unified hover's header already carries it, and
+ # three lanes over the full history made this the page's largest
+ # payload (about half a megabyte on gold).
+ cells.append(f"{head}
{s.label} {move} (z {zv:+.2f})")
+ out.append(cells)
+ return out
+
+
+def get_flow_state_strip(fig, df, row, col):
+ """Three lanes, one per opinion cohort, coloured by the sign of its flow.
+
+ Up or down past one sd in the polarity pair, inside one sd in grey, the warm-up
+ blank. The weeks whose state has the three cohorts all active and split (the
+ gold doc's mixed signs; `flows.DIVERGENT_FLOW_STATES`) get a tick in a thin lane
+ above the three. The plan asked for a line-only box around the week; in the
+ running app a facet column gives one to two pixels a week, where a one-pixel
+ outline is the whole cell and the box painted over the very lanes it was meant
+ to frame. BROAD_ACCUM and BROAD_LIQUID get no tick: everyone moving together is
+ visible in the lanes without help.
+
+ Lanes read top-down in the opinion order cotmetrics recorded, which is the
+ order the cohort rows above them take. No tick labels in a facet cell, for the
+ reason the flow rows give; the hover names the lane.
+
+ The state is caption vocabulary. Its pre-registered test failed, so the hover
+ and the page caption say so beside every name.
+ """
+ specs = _opinion_specs(df)
+ if len(specs) != 3 or const.FLOW_STATE not in df.columns:
+ return fig
+ z = []
+ for s in specs:
+ z.append([None if pd.isna(v) else float(v) for v in df[flows.flow_sign_col(s)]])
+ fig.add_trace(go.Heatmap(
+ x=df.index, y=[0, 1, 2], z=z,
+ text=_strip_hover_text(df, specs),
+ hovertemplate="%{text}",
+ zmin=-1, zmax=1, zmid=0,
+ colorscale=FLOW_SIGN_COLORSCALE,
+ xgap=0, ygap=1, hoverongaps=False, showscale=False,
+ name="Flow state", showlegend=False,
+ ), row=row, col=col)
+
+ split = df[const.FLOW_STATE].isin(flows.DIVERGENT_FLOW_STATES).to_numpy()
+ if split.any():
+ dates = df.index[split]
+ fig.add_trace(go.Scatter(
+ x=dates, y=[_SPLIT_LANE_Y] * len(dates), mode="markers",
+ marker=dict(symbol="line-ns", size=7,
+ line=dict(width=2, color=FLOW_SPLIT_COLOR)),
+ hoverinfo="skip", name="Split weeks", showlegend=False,
+ ), row=row, col=col)
+
+ fig.update_yaxes(row=row, col=col, type="linear", range=[2.6, -1.4],
+ fixedrange=True, showgrid=False, zeroline=False,
+ showticklabels=False, title=None)
+ return fig
+
+
+# --- the flow view -----------------------------------------------------------------
+# Three full-width panels on one time axis, the Cowork mockup (cotmetrics
+# docs/design/cot-flows.md, PR 3): price with open interest, every selected cohort's
+# flow with the counterparty row and the level markers, and every cohort's
+# positioning index. Price gets the tallest panel because it is what the other two
+# are read against.
+FLOW_VIEW_ROW_HEIGHTS = [0.36, 0.32, 0.32]
+FLOW_VIEW_SPECS = [[{"secondary_y": True}], [{"secondary_y": False}],
+ [{"secondary_y": False}]]
+FLOW_VIEW_BAND_OPACITY = 0.35
+
+
+def _level_weeks(df):
+ return df.attrs.get("flow_level_weeks") or df.attrs.get("lookback_weeks")
+
+
+def flow_view_titles(df):
+ weeks = _level_weeks(df)
+ span = f"{weeks}w range" if weeks else "range"
+ return [
+ "Price and Open Interest",
+ (f"Weekly Flow z ({const.FLOW_Z_WEEKS}w sd) "
+ f"\u25b2 left above {const.FLOW_LEVEL_HIGH}, "
+ f"\u25bc left below {const.FLOW_LEVEL_LOW} of the {span}"),
+ f"Positioning Index ({span})",
+ ]
+
+
+def build_flow_view(fig, df, series, lookback_header, palette):
+ """Fill a three-row figure made with FLOW_VIEW_SPECS and FLOW_VIEW_ROW_HEIGHTS.
+
+ The index panel is shaded above FLOW_LEVEL_HIGH and below FLOW_LEVEL_LOW, and
+ only here: those are the cutoffs the triangles in the panel above use, so the
+ shading is the key to the markers. The Positioning Index panel elsewhere on the
+ page stays unshaded for the reason its docstring gives, and nothing in this view
+ shades a setup gate.
+ """
+ if const.CLOSING_PRICE in df.columns:
+ add_trace_to_all(fig, df, const.CLOSING_PRICE, 1, 1, "Price",
+ palette[vc.CATEGORY_PRICE_SLOT], 1, opacity=0.95)
+ _primary_axis(fig, 1, 1, "Price",
+ y_range=_fit_range(df, [const.CLOSING_PRICE]))
+ if const.OPEN_INTEREST in df.columns:
+ add_trace_to_all(fig, df, const.OPEN_INTEREST, 1, 1, "Open Interest",
+ palette[vc.CATEGORY_OI_SLOT], 0, secondary=True,
+ opacity=0.6, dash="dot")
+ fig.update_yaxes(title="OI", row=1, col=1, secondary_y=True,
+ showgrid=False, zeroline=False, fixedrange=True,
+ range=_fit_range(df, [const.OPEN_INTEREST]))
+
+ rows = _flow_rows(df, series, lookback_header)
+ if rows:
+ _flow_heatmap(fig, df, rows, 2, 1, dense=True)
+
+ get_category_index_plot(fig, df, series, lookback_header, 3, 1, palette,
+ show_price=False, showlegend=True)
+ for y0, y1 in ((const.FLOW_LEVEL_HIGH, 100), (0, const.FLOW_LEVEL_LOW)):
+ fig.add_hrect(y0=y0, y1=y1, row=3, col=1, line_width=0, layer="below",
+ fillcolor=vc.GRID_COLOR, opacity=FLOW_VIEW_BAND_OPACITY)
+ return fig
+
+
# --- the page's plot vocabulary --------------------------------------------------
# id -> (label, builder, when the cell needs a secondary y-axis)
#
@@ -781,16 +1030,70 @@ def build_panel(plot_id, fig, df, series, lookback_header, row, col, palette,
# --- small multiples --------------------------------------------------------------
-def facet_context_rows(plots, show_price):
- """The non-category rows: price, and open interest when Net Positions is shown.
+# The two flow context rows, drawn only in the flow column (see build_facet_figure).
+FLOW_CONTEXT_ROWS = ("counterparty", "state")
+
+
+def counterparty_members(df):
+ """The counterparty's member specs whose flow the frame carries, in role order."""
+ roles = (df.attrs.get("flow_roles") or {}) if df is not None else {}
+ report = roles.get("report")
+ if report not in categories.REPORT_CHOICES:
+ return []
+ by_key = {s.key: s for s in categories.categories_for(report)}
+ return [by_key[k] for k in roles.get("counterparty") or ()
+ if k in by_key and flows.flow_col(by_key[k]) in df.columns]
- Both are context for the categories rather than categories themselves, and in the
- overlay view both ride a second y-axis. Two scales on one plot align arbitrarily,
- which invents a correlation the data does not contain, so here each gets its own
- row against the same x. Faceting has already produced the row structure, so this
- costs nothing.
+
+def needs_counterparty_row(df, series):
+ """A composite row in small multiples, unless it would repeat a row on the page.
+
+ Where the counterparty is one cohort (the report defaults, and most TFF markets)
+ and that cohort is already a row, a composite row is the same cells again two
+ rows further down; that cohort's own row is labelled as the counterparty instead
+ (`facet_row_label`). A multi-member composite (gold's Producer/Merchant plus
+ Swap Dealers), or a single member the checklist switched off, gets its row.
+ """
+ if df is None or flows.counterparty_flow_z_col() not in df.columns:
+ return False
+ members = counterparty_members(df)
+ shown = {s.key for s in series}
+ return not (len(members) == 1 and members[0].key in shown)
+
+
+def facet_row_label(df, s):
+ """A category row's axis title, naming it the counterparty where it is the one."""
+ members = counterparty_members(df)
+ if len(members) == 1 and members[0].key == s.key:
+ return f"{s.label}
(counterparty)"
+ return s.label
+
+
+def facet_context_rows(plots, show_price, frame, series=()):
+ """The non-category rows: the flow column's counterparty row and state strip,
+ then price, then open interest with Net Positions.
+
+ Price and open interest are context for the categories rather than categories
+ themselves, and in the overlay view both ride a second y-axis. Two scales on one
+ plot align arbitrarily, which invents a correlation the data does not contain,
+ so here each gets its own row against the same x. Faceting has already produced
+ the row structure, so this costs nothing.
+
+ With the flow panel selected, the counterparty is on the page: as its own
+ composite row directly under the cohorts it is the other side of, or, where it
+ is a single cohort already drawn, as that cohort's row (needs_counterparty_row).
+ The state strip follows it, on a state-eligible market only. Price and open
+ interest come last, so the bottom row carries every column's dates. `frame`
+ is required: the rows depend on the market, and a shape computed without it
+ would leave an empty row where the market has no composite or no state.
"""
rows = []
+ if "flow" in plots:
+ if needs_counterparty_row(frame, series):
+ rows.append(("counterparty", flows.counterparty_flow_z_col(),
+ "Counterparty", None))
+ if has_state_strip(frame):
+ rows.append(("state", const.FLOW_STATE, "Flow state", None))
if show_price:
rows.append(("price", const.CLOSING_PRICE, "Price", vc.CATEGORY_PRICE_SLOT))
if "net_pos" in plots:
@@ -798,19 +1101,19 @@ def facet_context_rows(plots, show_price):
return rows
-def facet_shape(plots, series, show_price):
+def facet_shape(plots, series, show_price, frame):
"""Grid shape for the faceted view: a row per category, a column per panel."""
- rows = len(series) + len(facet_context_rows(plots, show_price))
+ rows = len(series) + len(facet_context_rows(plots, show_price, frame, series))
return max(1, rows), max(1, len(plots))
-def facet_titles(plots, series, show_price):
+def facet_titles(plots, series, show_price, frame):
"""Panel names on the top row only; every other cell is unlabelled.
Category identity rides on the y-axis title of the first column instead, so it is
stated once per row rather than repeated in every cell.
"""
- rows, cols = facet_shape(plots, series, show_price)
+ rows, cols = facet_shape(plots, series, show_price, frame)
titles = []
for r in range(rows):
for c in range(cols):
@@ -818,8 +1121,8 @@ def facet_titles(plots, series, show_price):
return titles
-def facet_specs(plots, series, show_price):
- rows, cols = facet_shape(plots, series, show_price)
+def facet_specs(plots, series, show_price, frame):
+ rows, cols = facet_shape(plots, series, show_price, frame)
return [[{"secondary_y": False} for _ in range(cols)] for _ in range(rows)]
@@ -833,7 +1136,7 @@ def build_facet_figure(fig, df, series, plots, lookback_header, palette,
carries its own label, which is the relief the palette checks ask for on the two
shipped palettes whose lightened siblings sit near the chroma floor.
"""
- _, cols = facet_shape(plots, series, show_price)
+ _, cols = facet_shape(plots, series, show_price, df)
def label_axis(text, r, c):
fig.update_yaxes(title_text=text if c == 1 else "", row=r, col=c,
@@ -846,13 +1149,37 @@ def label_axis(text, r, c):
show_price=False, showlegend=False, y_range=y_range,
facet=True)
# Row identity, stated once, in text rather than by colour alone.
- label_axis(s.label, r, c)
+ label_axis(facet_row_label(df, s) if "flow" in plots else s.label, r, c)
- for i, (_, column, label, slot) in enumerate(
- facet_context_rows(plots, show_price)):
+ flow_col = plots.index("flow") + 1 if "flow" in plots else None
+ for i, (row_id, column, label, slot) in enumerate(
+ facet_context_rows(plots, show_price, df, series)):
if column not in df.columns:
continue
r = len(series) + 1 + i
+ if row_id in FLOW_CONTEXT_ROWS:
+ # Only the flow column has something to draw here. The other cells
+ # keep their x-axis (the bottom row carries the date labels for its
+ # column when price and open interest are off) and lose their y ticks
+ # and grid so they read as empty.
+ for c in range(1, cols + 1):
+ if c == flow_col:
+ draw = (get_counterparty_flow_row if row_id == "counterparty"
+ else get_flow_state_strip)
+ draw(fig, df, r, c)
+ else:
+ # plotly.js draws only the subplots some trace references, so
+ # a cell left with no trace loses its axes, the row label
+ # label_axis puts on column 1, and, in the bottom row, the
+ # column's dates. A trace with no points keeps the cell.
+ fig.add_trace(go.Scatter(
+ x=[df.index[0], df.index[-1]], y=[None, None],
+ mode="markers", hoverinfo="skip", showlegend=False,
+ name=""), row=r, col=c)
+ fig.update_yaxes(row=r, col=c, showticklabels=False,
+ showgrid=False, zeroline=False, fixedrange=True)
+ label_axis(label, r, c)
+ continue
for c in range(1, cols + 1):
add_trace_to_all(fig, df, column, r, c, label, palette[slot], 0,
opacity=0.9)
diff --git a/src/components/flow_copy.py b/src/components/flow_copy.py
new file mode 100644
index 0000000..0731db8
--- /dev/null
+++ b/src/components/flow_copy.py
@@ -0,0 +1,263 @@
+"""The week in words: the caption under the Weekly Flow panel on /categories.
+
+One report week, told as sentences: which week, each opinion cohort's net change
+with both legs and its z, the counterparty always (named per market from the roles
+cotmetrics recorded in the frame's attrs), the cohorts on neither side, the
+sum-to-zero line, and the state as a vocabulary label. Everything is read from the
+frame `CotIndexer.get_category_data` returns; nothing here computes a flow, a z, a
+level or a state, and nothing reads the store, so the copy is testable under CI's
+empty COTDATA_STORE.
+
+Copy rules, held by tests (cotmetrics docs/design/cot-flows.md, PR 2 and PR 3):
+
+* Never the Home board's ranking words ("mover", "biggest move", "unusual"): a flow
+ z is contracts against one cohort's own history, not an index-point change of the
+ Legacy Commercial leg.
+* Nothing about what comes next. The state's pre-registered test failed the crucible
+ gauntlet on 2026-09-26, and the level cutoffs behind the markers were descriptive
+ and on gold alone, so no sentence may say what follows a state or a flow from an
+ extreme. "Signal" appears only as "not a signal".
+* The weekday is read from the date, never written as "Tuesday", and no release date
+ is stated: the CFTC moves both on holiday weeks, and the resolved release date
+ lives in cotdata's vintage store, which this module does not read.
+"""
+
+import cotmetrics.categories as categories
+import cotmetrics.constants as const
+import cotmetrics.flow_roles as flow_roles
+import cotmetrics.flows as flows
+import pandas as pd
+
+VOCABULARY_NOTE = "a vocabulary label, not a signal"
+
+
+def _contracts(v):
+ return f"{abs(v):,.0f}"
+
+
+def _net_phrase(dnet):
+ if pd.isna(dnet):
+ return None
+ if dnet > 0:
+ return f"net bought {_contracts(dnet)} contracts"
+ if dnet < 0:
+ return f"net sold {_contracts(dnet)} contracts"
+ return "left its net position unchanged"
+
+
+def _signed(v):
+ return "n/a" if pd.isna(v) else f"{v:+,.0f}"
+
+
+def _value(df, column, i):
+ if column is None or column not in df.columns:
+ return None
+ v = df[column].iloc[i]
+ return None if pd.isna(v) else v
+
+
+def _when(date):
+ if hasattr(date, "strftime"):
+ return f"{date.strftime('%A')} {date.strftime('%Y-%m-%d')}"
+ return str(date)
+
+
+def _weeks(df):
+ return df.attrs.get("flow_level_weeks") or df.attrs.get("lookback_weeks")
+
+
+def cohort_sentence(df, spec, i, lookback_header):
+ """One cohort's week: net change with both legs, z, thin note, marked level."""
+ dnet = _value(df, flows.flow_col(spec), i)
+ phrase = _net_phrase(dnet)
+ if phrase is None:
+ return (f"{spec.label}: no reading this week (the report has a gap or a "
+ f"contract switch here, so the change is not a week's flow).")
+ dlong = _value(df, flows.flow_long_col(spec), i)
+ dshort = _value(df, flows.flow_short_col(spec), i)
+ text = f"{spec.label} {phrase}"
+ if dlong is not None and dshort is not None:
+ text += f" (longs {_signed(dlong)}, shorts {_signed(dshort)})"
+ z = _value(df, flows.flow_z_col(spec), i)
+ if z is None:
+ text += (f", z not readable (under {const.FLOW_Z_MIN_PERIODS} weeks of "
+ f"history, or no week-to-week variation)")
+ else:
+ # Two decimals, as the hover prints: at one, 1.04 (active) and 0.96 (not)
+ # both read "+1.0". "Inside one sd" is cotmetrics' sign, not a comparison
+ # made here, and only the opinion cohorts carry one.
+ text += f", z {z:+.2f} against its own {const.FLOW_Z_WEEKS}-week sd"
+ if _value(df, flows.flow_sign_col(spec), i) == 0:
+ text += ", inside one sd"
+ if _value(df, flows.flow_thin_col(spec), i):
+ text += (f"; thin, its typical week is under {const.FLOW_MIN_STD_CONTRACTS} "
+ f"contracts, so read the count rather than the z")
+ if lookback_header is not None:
+ mark = _value(df, flows.flow_level_mark_col(spec, lookback_header), i)
+ level = _value(df, flows.flow_from_level_col(spec, lookback_header), i)
+ if mark and level is not None:
+ # Worded from the mark, with the level to one decimal, so 19.6 cannot
+ # print as "20, the bottom" against a strict below-20 rule.
+ where = (f"above {const.FLOW_LEVEL_HIGH}" if mark > 0
+ else f"below {const.FLOW_LEVEL_LOW}")
+ text += f", leaving level {level:.1f} ({where}) of its {_span(df)}"
+ return text + "."
+
+
+def _span(df):
+ weeks = _weeks(df)
+ return f"{weeks}-week range" if weeks else "lookback range"
+
+
+def marker_key(df):
+ """What the triangles mean, once, in words."""
+ return (f"A triangle marks an active week (beyond one sd) that left the top "
+ f"(up, the index above {const.FLOW_LEVEL_HIGH} the week before) or the "
+ f"bottom (down, below {const.FLOW_LEVEL_LOW}) of the cohort's "
+ f"{_span(df)}.")
+
+
+def _by_key(report):
+ return {s.key: s for s in categories.categories_for(report)}
+
+
+def _present(df, keys, report):
+ by_key = _by_key(report)
+ return [by_key[k] for k in keys
+ if k in by_key and flows.flow_col(by_key[k]) in df.columns]
+
+
+def counterparty_sentence(df, roles, report, i):
+ members = _present(df, roles.get("counterparty") or (), report)
+ column = flows.counterparty_flow_col()
+ if not members or column not in df.columns:
+ return ("No counterparty composite on this market: none of its measured "
+ "members is in the report.")
+ names = " + ".join(s.label for s in members)
+ source = roles.get("source") or ""
+ label = categories.REPORT_LABELS.get(report, report)
+ if source == flow_roles.SOURCE_MEASURED:
+ why = "measured as this market's counterparty"
+ elif source == flow_roles.SOURCE_UNSTABLE:
+ why = (f"the {label} default, since this market's own roles did not hold "
+ f"steady enough to measure")
+ else:
+ why = f"the {label} default; this market's roles were not measured"
+ phrase = _net_phrase(_value(df, column, i))
+ if phrase is None:
+ return f"On the other side, {names} ({why}): no reading this week."
+ text = f"On the other side, {names} ({why}) {phrase}"
+ z = _value(df, flows.counterparty_flow_z_col(), i)
+ if z is not None:
+ text += f", z {z:+.2f}"
+ # Where the measured set shares a cohort with the opinion set (Other
+ # Reportable on silver, copper, orange juice), that cohort's flow is in the
+ # sentence above and in this one, so the printed figures do not sum to zero.
+ # Say so rather than let the sum-to-zero line below contradict them.
+ opinion = set(roles.get("opinion") or ())
+ both = [s.label for s in members if s.key in opinion]
+ if both:
+ text += (f". {' and '.join(both)} {'is' if len(both) == 1 else 'are'} also "
+ f"one of the cohorts above, so {'its' if len(both) == 1 else 'their'} "
+ f"flow is counted on both sides here")
+ return text + "."
+
+
+def neither_side_sentence(df, roles, report, i):
+ keys = [k for k in (tuple(roles.get("neutral") or ())
+ + tuple(roles.get("inert") or ())
+ + tuple(roles.get("residual") or ()))
+ if k not in (roles.get("counterparty") or ())
+ and k not in (roles.get("opinion") or ())]
+ specs = _present(df, keys, report)
+ if not specs:
+ return None
+ parts = []
+ for s in specs:
+ v = _value(df, flows.flow_col(s), i)
+ parts.append(f"{s.label} {_signed(v)} contracts net" if v is not None
+ else f"{s.label} no reading this week")
+ return f"On neither side: {', '.join(parts)}."
+
+
+def sum_to_zero_sentence(df, report):
+ specs = categories.categories_for(report)
+ if not all(flows.flow_col(s) in df.columns for s in specs):
+ return None
+ return ("Every contract bought was sold by someone, so the net changes of all "
+ "the cohorts, each counted once, sum to zero each week.")
+
+
+def state_sentence(df, roles, report, i):
+ if not roles.get("state_eligible", False) or const.FLOW_STATE not in df.columns:
+ return ("No state is named on this market: its cohorts do not split into "
+ "opinion and counterparty the way the state assumes, so the cells "
+ "are drawn and not labelled.")
+ state = df[const.FLOW_STATE].iloc[i]
+ if state is None or (isinstance(state, float) and pd.isna(state)):
+ return "No state this week: an opinion cohort's z is not readable yet."
+ opinion = _present(df, roles.get("opinion") or (), report)
+ if state == flows.FLOW_STATE_QUIET:
+ return (f"State: {state}, no opinion cohort moved beyond one sd "
+ f"({VOCABULARY_NOTE}).")
+ if state == flows.FLOW_STATE_PARTIAL:
+ n = _value(df, const.FLOW_N_ACTIVE, i)
+ count = f"{int(n)} of the {len(opinion)}" if n is not None else "some of the"
+ return (f"State: {state}, {count} opinion cohorts moved beyond one sd "
+ f"({VOCABULARY_NOTE}).")
+ moves = []
+ for s in opinion:
+ sign = _value(df, flows.flow_sign_col(s), i)
+ if sign:
+ moves.append(f"{s.label} {'buying' if sign > 0 else 'selling'}")
+ split = ("the opinion cohorts split" if state in flows.DIVERGENT_FLOW_STATES
+ else "the opinion cohorts moved together")
+ return f"State: {state} ({'; '.join(moves)}; {split}), {VOCABULARY_NOTE}."
+
+
+def latest_row(df):
+ """The last row where any opinion cohort's flow is readable, else the last row."""
+ roles = df.attrs.get("flow_roles") or {}
+ report = roles.get("report") or df.attrs.get("report")
+ cols = [flows.flow_col(s) for s in _present(df, roles.get("opinion") or (), report)]
+ if cols:
+ readable = df[cols].notna().any(axis=1).to_numpy()
+ hits = readable.nonzero()[0]
+ if len(hits):
+ return int(hits[-1])
+ return len(df) - 1
+
+
+def week_in_words(df, lookback_header, i=None):
+ """The caption as a list of sentences, for one row of the category frame.
+
+ `i` is a positional row; None takes `latest_row`. Returns [] when the frame has
+ no flow columns or no roles, so the page can call it unconditionally.
+ """
+ if df is None or df.empty:
+ return []
+ roles = df.attrs.get("flow_roles") or {}
+ report = roles.get("report") or df.attrs.get("report")
+ if report not in categories.REPORT_CHOICES:
+ return []
+ opinion = _present(df, roles.get("opinion") or (), report)
+ if not opinion:
+ return []
+ latest = i is None
+ i = latest_row(df) if latest else i
+ out = [f"Positions as of {_when(df.index[i])} against the report before: each "
+ f"cohort's net change in contracts, and its z against its own "
+ f"{const.FLOW_Z_WEEKS}-week sd of weekly changes."]
+ if latest and i < len(df) - 1:
+ out.append(f"The newest report, {_when(df.index[-1])}, has no reading for "
+ f"these cohorts (a gap or a contract switch), so this is the "
+ f"week before it.")
+ out += [cohort_sentence(df, s, i, lookback_header) for s in opinion]
+ out.append(counterparty_sentence(df, roles, report, i))
+ for sentence in (neither_side_sentence(df, roles, report, i),
+ sum_to_zero_sentence(df, report),
+ state_sentence(df, roles, report, i),
+ marker_key(df) if lookback_header is not None else None):
+ if sentence:
+ out.append(sentence)
+ return out
diff --git a/src/components/plot_layout.py b/src/components/plot_layout.py
index 529919e..75ae09f 100644
--- a/src/components/plot_layout.py
+++ b/src/components/plot_layout.py
@@ -113,7 +113,8 @@ def get_make_subplots_for_facets(rows, cols, titles, specs):
)
-def get_make_subplots_for_plots(rows, cols, titles, specs, shared_xaxes=True):
+def get_make_subplots_for_plots(rows, cols, titles, specs, shared_xaxes=True,
+ row_heights=None):
if rows > 1:
# vertical_spacing is a fraction of the plot area, not of the figure, so the
# denominator has to be the plot area or the gap comes out short.
@@ -128,7 +129,8 @@ def get_make_subplots_for_plots(rows, cols, titles, specs, shared_xaxes=True):
vertical_spacing=v_spacing,
horizontal_spacing=0.08,
subplot_titles=titles,
- specs=specs
+ specs=specs,
+ row_heights=row_heights,
)
return fig
diff --git a/src/pages/analytics/categories.py b/src/pages/analytics/categories.py
index 7d96b2c..760ba7a 100644
--- a/src/pages/analytics/categories.py
+++ b/src/pages/analytics/categories.py
@@ -25,7 +25,7 @@
import components.plot_layout as layout_helpers
import viz_config
import viz_constants as vc
-from components import config_fold, controls
+from components import config_fold, controls, flow_copy
dash.register_page(
__name__,
@@ -337,6 +337,17 @@ def set_default_columns(pathname, current_val):
return new_val
+@callback(
+ Output('categories_plot_selector', 'disabled'),
+ Output('categories_columns_selector', 'disabled'),
+ Input('categories_layout_selector', 'value'),
+)
+def lock_controls_for_flow_view(layout_mode):
+ """The Flow view is a fixed figure; say so by greying out what it ignores."""
+ flow_view = layout_mode == vc.LAYOUT_FLOW
+ return flow_view, flow_view
+
+
@callback(
Output('categories_stack', 'children'),
Input('session_palette_theme_asset_store', 'data'),
@@ -373,15 +384,29 @@ def render_category_stack(palette_name, asset, report, selected_categories,
show_price = True
lookback_header = df.attrs.get("lookback_header", " Custom")
faceted = layout_mode == vc.LAYOUT_FACET
-
- if faceted:
+ flow_view = layout_mode == vc.LAYOUT_FLOW
+
+ if flow_view:
+ # A fixed three-panel figure, so the plot selector and Cols do not apply
+ # (the callback below greys them out).
+ num_rows, num_cols = 3, 1
+ fig = layout_helpers.get_make_subplots_for_plots(
+ num_rows, num_cols, ct.flow_view_titles(df), ct.FLOW_VIEW_SPECS,
+ row_heights=ct.FLOW_VIEW_ROW_HEIGHTS)
+ fig = ct.build_flow_view(fig, df, series, lookback_header, palette)
+ plots = ["flow"]
+ height = None
+ show_legend = True
+ elif faceted:
# Rows are categories and columns are panels, so the Cols control does not
# apply: the column count is the number of panels selected.
- num_rows, num_cols = ct.facet_shape(plots, series, show_price)
+ # The frame goes in so a market without a counterparty composite or a
+ # state (TFF equities, rates, crypto) gets no empty flow context row.
+ num_rows, num_cols = ct.facet_shape(plots, series, show_price, frame=df)
fig = layout_helpers.get_make_subplots_for_facets(
num_rows, num_cols,
- ct.facet_titles(plots, series, show_price),
- ct.facet_specs(plots, series, show_price))
+ ct.facet_titles(plots, series, show_price, frame=df),
+ ct.facet_specs(plots, series, show_price, frame=df))
fig = ct.build_facet_figure(fig, df, series, plots, lookback_header, palette,
show_price=show_price)
height = layout_helpers.get_facet_figure_height(num_rows, num_cols)
@@ -411,7 +436,9 @@ def render_category_stack(palette_name, asset, report, selected_categories,
show_legend = True
fig = layout_helpers.get_update_xaxes_for_plots(fig, df)
- if faceted:
+ if faceted or flow_view:
+ # Dates under the bottom panel only; the flow view's three panels share
+ # one time axis, as small multiples do.
fig = layout_helpers.hide_inner_facet_xlabels(fig, num_rows, num_cols)
weeks = df.attrs.get("lookback_weeks")
@@ -422,13 +449,34 @@ def render_category_stack(palette_name, asset, report, selected_categories,
if not show_legend:
fig.update_layout(showlegend=False)
- return dcc.Graph(figure=fig,
- id='categories_main_graph',
- config={
- 'scrollZoom': False,
- 'doubleClick': 'reset',
- 'displayModeBar': not app_utils.is_mobile(),
- 'modeBarButtonsToRemove': ['pan2d', 'select2d', 'lasso2d'],
- 'displaylogo': False,
- 'responsive': True},
- style={'width': '100%'})
+ graph = dcc.Graph(figure=fig,
+ id='categories_main_graph',
+ config={
+ 'scrollZoom': False,
+ 'doubleClick': 'reset',
+ 'displayModeBar': not app_utils.is_mobile(),
+ 'modeBarButtonsToRemove': ['pan2d', 'select2d', 'lasso2d'],
+ 'displaylogo': False,
+ 'responsive': True},
+ style={'width': '100%'})
+ if "flow" not in plots:
+ return graph
+ return [graph, _flow_caption(df, lookback_header)]
+
+
+def _flow_caption(df, lookback_header):
+ """The latest week in words under the graph, when the flow panel is shown.
+
+ Written by components.flow_copy from the frame alone; this page adds only the
+ markup.
+ """
+ sentences = flow_copy.week_in_words(df, lookback_header)
+ if not sentences:
+ return html.Div()
+ return html.Div(
+ [html.P(sentences[0], style={'marginBottom': '0.35rem',
+ 'color': vc.BRIGHTER_TEXT_COLOR})]
+ + [html.P(s, style={'marginBottom': '0.2rem'}) for s in sentences[1:]],
+ id='categories_flow_caption',
+ style={'fontSize': '0.8rem', 'color': vc.TEXT_COLOR, 'maxWidth': '900px',
+ 'margin': '0.5rem auto 1rem', 'padding': '0 16px'})
diff --git a/src/viz_constants.py b/src/viz_constants.py
index 2041a95..2f73bdc 100644
--- a/src/viz_constants.py
+++ b/src/viz_constants.py
@@ -291,10 +291,15 @@ def positioning_tooltip(state, model, is_equity=False):
# How the page arranges the categories.
LAYOUT_OVERLAY = "overlay"
LAYOUT_FACET = "facet"
-LAYOUT_CHOICES = (LAYOUT_OVERLAY, LAYOUT_FACET)
+# /categories only: price with open interest, the flow heatmap and the positioning
+# index as three full-width panels on one time axis (cotmetrics docs/design/
+# cot-flows.md, PR 3, the Cowork mockup). Ignores the plot selector.
+LAYOUT_FLOW = "flow"
+LAYOUT_CHOICES = (LAYOUT_OVERLAY, LAYOUT_FACET, LAYOUT_FLOW)
LAYOUT_LABELS = {
LAYOUT_OVERLAY: "Overlay",
LAYOUT_FACET: "Small multiples",
+ LAYOUT_FLOW: "Flow view",
}
CATEGORY_PRICE_SLOT = 3
CATEGORY_OI_SLOT = 4
diff --git a/tests/test_category_traces.py b/tests/test_category_traces.py
index ef4c398..969187a 100644
--- a/tests/test_category_traces.py
+++ b/tests/test_category_traces.py
@@ -11,6 +11,7 @@
import pandas as pd
import pytest
from cotmetrics import categories as cot_categories
+from cotmetrics import constants as cm_const
from cotmetrics import flows
import components.category_traces as ct
@@ -71,6 +72,22 @@ def _frame(report, n=80, seed=3):
dtype="boolean")
df[flows.counterparty_flow_col()] = rng.normal(0, 5000, n)
df[flows.counterparty_flow_z_col()] = _warmup(rng.normal(0, 1, n))
+ # The level pair and the state, through cotmetrics' own functions so the frame
+ # is shaped exactly as build_flow_frame shapes it.
+ for spec in specs:
+ level = df[cot_categories.index_col(spec, HEADER)].shift(1)
+ df[flows.flow_from_level_col(spec, HEADER)] = level
+ df[flows.flow_level_mark_col(spec, HEADER)] = flows.level_marks(
+ df[flows.flow_z_col(spec)], level)
+ opinion = specs[2:5]
+ signs = [flows.flow_signs(df[flows.flow_z_col(s)]) for s in opinion]
+ for spec, sign in zip(opinion, signs):
+ df[flows.flow_sign_col(spec)] = sign
+ df[cm_const.FLOW_N_ACTIVE] = sum((x != 0).astype("Int64") for x in signs)
+ df[cm_const.FLOW_STATE] = flows.flow_state(*signs, report=report)
+ df.attrs["lookback_header"] = HEADER
+ df.attrs["lookback_weeks"] = 52
+ df.attrs["flow_level_weeks"] = 52
df.attrs["report"] = report
df.attrs["flow_roles"] = {
"report": report,
@@ -134,6 +151,8 @@ def test_every_panel_draws_a_trace_per_category(report, plot_id):
if plot_id == "flow":
# One heatmap carries every category as a row, plus the counterparty. The
# composite's tick label is the bare word; its members are in the hover.
+ # The level markers ride beside it as one annotation trace.
+ drawn = [t for t in drawn if t.name != "Level marks"]
assert len(drawn) == 1 and drawn[0].type == "heatmap"
assert list(drawn[0].y) == [s.label for s in series] + ["Counterparty"]
assert list(drawn[0].x) == list(df.index)
@@ -299,10 +318,10 @@ def test_index_panel_keeps_its_fixed_scale():
def _facet(df, series, plots, show_price=True):
- rows, cols = ct.facet_shape(plots, series, show_price)
+ rows, cols = ct.facet_shape(plots, series, show_price, frame=df)
fig = layout_helpers.get_make_subplots_for_facets(
- rows, cols, ct.facet_titles(plots, series, show_price),
- ct.facet_specs(plots, series, show_price))
+ rows, cols, ct.facet_titles(plots, series, show_price, frame=df),
+ ct.facet_specs(plots, series, show_price, frame=df))
return ct.build_facet_figure(fig, df, series, plots, HEADER, PALETTE,
show_price=show_price), rows, cols
@@ -319,12 +338,18 @@ def test_facet_gives_each_category_its_own_row(report):
plots = ["net_pos", "index", "flow"]
fig, rows, cols = _facet(df, series, plots)
- # 5 categories + a price row + an open-interest row (net_pos is selected).
- assert rows == len(series) + 2
+ # 5 categories + a price row + an open-interest row (net_pos is selected)
+ # + the flow column's counterparty row and state strip (flow is selected).
+ assert rows == len(series) + 4
assert cols == len(plots)
+ # One series per cell. The level markers and the split-week ticks annotate
+ # the series in their cell rather than being series of their own.
+ annotations = {"Level marks", "Split weeks"}
per_cell = {}
for t in fig.data:
+ if t.name in annotations:
+ continue
per_cell.setdefault(t.yaxis, []).append(t)
assert all(len(v) == 1 for v in per_cell.values()), \
"a faceted cell must hold exactly one series"
@@ -712,12 +737,16 @@ def test_flow_heatmap_skips_a_category_missing_flow_columns():
@pytest.mark.parametrize("report", list(cot_categories.REPORT_CHOICES))
def test_flow_facet_is_one_row_heatmap_per_category(report):
- """In small multiples each cell is a one-row heatmap; the composite is not a row."""
+ """In small multiples each category cell is a one-row heatmap, composite apart."""
df = _frame(report)
series = ct.category_series(report, None, PALETTE, frame=df)
fig, rows, cols = _facet(df, series, ["flow"], show_price=False)
- maps = [t for t in fig.data if t.type == "heatmap"]
- assert len(maps) == len(series)
+ # The category rows only: the counterparty row and the state strip under them
+ # are context rows, tested on their own below.
+ all_maps = [t for t in fig.data if t.type == "heatmap"]
+ # The categories, then the composite row and the strip; nothing else.
+ assert len(all_maps) == len(series) + 2
+ maps = all_maps[:len(series)]
assert [list(m.y) for m in maps] == [[s.label] for s in series]
assert not any("Counterparty" in label for m in maps for label in m.y)
assert all(m.showscale is False for m in maps)
@@ -790,3 +819,283 @@ def test_flow_heatmap_has_no_colorbar_and_no_price():
entries = {t.name for t in fig.data if t.x is not None and len(t.x)
and t.x[0] is None}
assert entries == {s.label for s in series}
+
+
+# --- PR 3: level markers, the counterparty context row, the state strip ----------
+
+def _marks(fig):
+ got = [t for t in fig.data if t.name == "Level marks"]
+ assert len(got) <= 1
+ return got[0] if got else None
+
+
+def test_level_markers_sit_on_exactly_the_marked_cells():
+ """A triangle per cell whose active flow left an extreme; up for the top.
+
+ The marks are cotmetrics' (flows.flow_level_mark_col), so the view draws what
+ the frame says and computes nothing. The composite has no range index of its
+ own and so never carries a marker.
+ """
+ report = cot_categories.REPORT_DISAGG
+ fig, df, series = _flow_figure(report)
+ marks = _marks(fig)
+ want = []
+ for s in series:
+ m = df[flows.flow_level_mark_col(s.spec, HEADER)]
+ for date, v in zip(df.index, m):
+ if not pd.isna(v) and v != 0:
+ want.append((date, s.label, "triangle-up" if v > 0 else "triangle-down"))
+ assert want, "the test frame must carry some marks"
+ got = list(zip(marks.x, marks.y, marks.marker.symbol))
+ assert sorted(got) == sorted(want)
+ assert "Counterparty" not in set(marks.y)
+ assert marks.hoverinfo == "skip"
+ assert marks.showlegend is False
+
+
+def test_marked_cells_name_their_level_and_unmarked_cells_do_not():
+ report = cot_categories.REPORT_DISAGG
+ fig, df, series = _flow_figure(report)
+ hm = _heatmap(fig)
+ for r, s in enumerate(series):
+ mark = df[flows.flow_level_mark_col(s.spec, HEADER)]
+ level = df[flows.flow_from_level_col(s.spec, HEADER)]
+ for i, text in enumerate(hm.text[r]):
+ if text is None:
+ continue
+ marked = not pd.isna(mark.iloc[i]) and mark.iloc[i] != 0
+ if marked:
+ assert f"from level {level.iloc[i]:.1f} of the 52-week range" in text
+ else:
+ assert "from level" not in text
+
+
+def test_no_markers_when_the_frame_has_no_level_columns():
+ report = cot_categories.REPORT_DISAGG
+ df = _frame(report)
+ df = df.drop(columns=[c for c in df.columns
+ if cm_const.FLOW_LEVEL_MARK in c or cm_const.FLOW_FROM_LEVEL in c])
+ fig, _, _ = _flow_figure(report, df=df)
+ assert _marks(fig) is None
+ assert not any("from level" in (c or "") for row in _heatmap(fig).text for c in row)
+
+
+def _facet_flow(df=None, plots=("flow",), show_price=True):
+ report = cot_categories.REPORT_DISAGG
+ df = _frame(report) if df is None else df
+ series = ct.category_series(report, None, PALETTE, frame=df)
+ fig, rows, cols = _facet(df, series, list(plots), show_price=show_price)
+ return fig, df, series, rows, cols
+
+
+def _strip(fig):
+ got = [t for t in fig.data if t.name == "Flow state"]
+ assert len(got) <= 1
+ return got[0] if got else None
+
+
+def test_facet_flow_puts_the_counterparty_and_the_strip_right_under_the_cohorts():
+ fig, df, series, rows, cols = _facet_flow()
+ assert rows == len(series) + 3 # counterparty, strip, price
+ names = [r[0] for r in ct.facet_context_rows(["flow"], True, df, series)]
+ assert names == ["counterparty", "state", "price"]
+
+ composite = [t for t in fig.data if t.type == "heatmap" and list(t.y) == ["Counterparty"]]
+ assert len(composite) == 1
+ assert composite[0].text[-1][-1].startswith(_counterparty_hover_label(
+ cot_categories.REPORT_DISAGG))
+
+ strip = _strip(fig)
+ assert list(strip.y) == [0, 1, 2]
+ values = {v for lane in strip.z for v in lane if v is not None}
+ assert values <= {-1.0, 0.0, 1.0}
+ # Lanes follow the opinion order cotmetrics recorded, and warm-up weeks are blank.
+ opinion = df.attrs["flow_roles"]["opinion"]
+ by_key = {s.key: s for s in cot_categories.categories_for(cot_categories.REPORT_DISAGG)}
+ for lane, key in zip(strip.z, opinion):
+ sign = df[flows.flow_sign_col(by_key[key])]
+ assert [None if pd.isna(v) else float(v) for v in sign] == list(lane)
+ assert all(v is None for lane in strip.z for v in lane[:WARMUP])
+ assert strip.showscale is False and strip.hoverongaps is False
+
+
+def test_state_strip_ticks_only_the_split_weeks():
+ """Mixed all-active signs get a tick; moving together, PARTIAL and QUIET do not."""
+ report = cot_categories.REPORT_DISAGG
+ df = _frame(report)
+ state = pd.Series([None] * len(df), index=df.index, dtype=object)
+ state.iloc[WARMUP:] = flows.FLOW_STATE_QUIET
+ state.iloc[40] = "VALUE_ACCUM"
+ state.iloc[41] = "BROAD_ACCUM"
+ state.iloc[42] = flows.FLOW_STATE_PARTIAL
+ state.iloc[50] = "MM_SELL_OTHERS_BUY"
+ df[cm_const.FLOW_STATE] = state
+ fig, _, _, _, _ = _facet_flow(df=df)
+ ticks = [t for t in fig.data if t.name == "Split weeks"]
+ assert len(ticks) == 1
+ assert [pd.Timestamp(x) for x in ticks[0].x] == [df.index[40], df.index[50]]
+ # Above the three lanes, never over them.
+ assert all(y < -0.5 for y in ticks[0].y)
+ assert ticks[0].hoverinfo == "skip"
+
+
+def test_state_strip_hover_says_label_not_signal():
+ fig, df, _, _, _ = _facet_flow()
+ strip = _strip(fig)
+ texts = [c for lane in strip.text for c in lane if c]
+ assert texts
+ named = [t for t in texts if not t.startswith("no state")]
+ assert named and all("a vocabulary label, not a signal" in t for t in named)
+ # The date is the unified hover's header, not repeated in every lane.
+ assert not any("positions as of" in t for t in texts)
+
+
+def test_ineligible_market_gets_no_strip_row_and_no_empty_row():
+ """TFF equities, rates and crypto carry z cells and no state (ADR-0005)."""
+ df = _frame(cot_categories.REPORT_DISAGG).drop(columns=[cm_const.FLOW_STATE])
+ fig, _, series, rows, _ = _facet_flow(df=df)
+ assert rows == len(series) + 2 # price and counterparty only
+ assert _strip(fig) is None
+ assert not ct.has_state_strip(df)
+
+
+def test_flow_context_rows_leave_other_columns_empty_but_keep_their_dates():
+ """Only the flow column has a composite and a state to draw.
+
+ The other cells in those rows stay empty, and the bottom row still carries the
+ date labels for its column, because hiding a bottom cell's x-axis would take
+ the whole column's dates with it.
+ """
+ fig, df, series, rows, cols = _facet_flow(plots=("net_pos", "flow"))
+ assert cols == 2
+ context = ct.facet_context_rows(["net_pos", "flow"], True, df, series)
+ first = len(series) + 1
+ rows_of = {row_id: first + i for i, (row_id, *_rest) in enumerate(context)}
+ for row_id in ("counterparty", "state"):
+ r = rows_of[row_id]
+ refs = fig.get_subplot(r, 1)
+ axis_name = refs.yaxis.plotly_name.replace("yaxis", "y")
+ in_col1 = [t for t in fig.data if t.yaxis == axis_name]
+ # Exactly one trace, with no points: plotly.js drops a subplot no trace
+ # references, and with it the row label and the column's dates.
+ assert len(in_col1) == 1, row_id
+ assert all(v is None for v in in_col1[0].y)
+ assert in_col1[0].hoverinfo == "skip"
+ assert refs.yaxis.showticklabels is False
+ assert refs.yaxis.title.text in {"Counterparty", "Flow state"}
+ bottom_x = fig.get_subplot(rows, 1).xaxis
+ assert bottom_x.showticklabels is not False
+
+
+def test_pr3_copy_never_ranks_or_forecasts():
+ """The marker and strip copy obeys the panel's copy rule and says nothing about
+ what follows a flow or a state."""
+ banned = ("mover", "biggest move", "unusual", "next", "follow", "expect",
+ "forecast", "predict", "bullish", "bearish", "likely", "edge")
+ fig, _, _, _, _ = _facet_flow()
+ texts = [c for lane in _strip(fig).text for c in lane if c]
+ flow_fig, _, _ = _flow_figure(cot_categories.REPORT_DISAGG)
+ texts += [c for row in _heatmap(flow_fig).text for c in row if c]
+ # What a reader sees: hover text and trace names. Docstrings are left out on
+ # purpose, since they have to name the rule to explain it.
+ texts += [t.name for t in list(fig.data) + list(flow_fig.data) if t.name]
+ for text in texts:
+ low = text.lower().replace("not a signal", "")
+ assert not any(b in low for b in banned), text
+ assert "signal" not in low, text
+
+
+def _single_member(df, key):
+ df.attrs["flow_roles"] = {**df.attrs["flow_roles"], "counterparty": (key,)}
+ return df
+
+
+def test_a_one_cohort_counterparty_is_labelled_on_its_own_row_not_repeated():
+ """DOW's counterparty is Dealer/Intermediary alone: a composite row would be the
+ first row again, two rows further down. Its own row says what it is instead."""
+ report = cot_categories.REPORT_DISAGG
+ df = _single_member(_frame(report), "producer_merchant")
+ fig, _, series, rows, _ = _facet_flow(df=df)
+ assert rows == len(series) + 2 # strip and price, no composite row
+ assert not [t for t in fig.data if t.type == "heatmap" and list(t.y) == ["Counterparty"]]
+ titles = [fig.layout["yaxis" + ("" if r == 1 else str(r))].title.text
+ for r in range(1, len(series) + 1)]
+ assert titles[0] == "Producer/Merchant
(counterparty)"
+ assert all("(counterparty)" not in t for t in titles[1:])
+
+
+def test_a_one_cohort_counterparty_switched_off_still_gets_its_row():
+ report = cot_categories.REPORT_DISAGG
+ df = _single_member(_frame(report), "producer_merchant")
+ keys = [s.key for s in cot_categories.categories_for(report)][1:]
+ series = ct.category_series(report, keys, PALETTE, frame=df)
+ assert ct.needs_counterparty_row(df, series)
+ fig, rows, _ = _facet(df, series, ["flow"], show_price=False)
+ assert [t for t in fig.data if t.type == "heatmap" and list(t.y) == ["Counterparty"]]
+
+
+def test_facet_rows_are_labelled_only_when_flow_is_shown():
+ report = cot_categories.REPORT_DISAGG
+ df = _single_member(_frame(report), "producer_merchant")
+ series = ct.category_series(report, None, PALETTE, frame=df)
+ fig, _, _ = _facet(df, series, ["net_pos"])
+ assert fig.layout.yaxis.title.text == "Producer/Merchant"
+
+
+# --- the flow view (the Cowork mockup) -------------------------------------------
+
+def _flow_view(report=cot_categories.REPORT_DISAGG, df=None):
+ df = _frame(report) if df is None else df
+ series = ct.category_series(report, None, PALETTE, frame=df)
+ fig = layout_helpers.get_make_subplots_for_plots(
+ 3, 1, ct.flow_view_titles(df), ct.FLOW_VIEW_SPECS,
+ row_heights=ct.FLOW_VIEW_ROW_HEIGHTS)
+ return ct.build_flow_view(fig, df, series, HEADER, PALETTE), df, series
+
+
+def test_flow_view_is_price_then_flow_then_index_on_one_time_axis():
+ fig, df, series = _flow_view()
+ by_axis = {}
+ for t in _named_traces(fig):
+ by_axis.setdefault(t.yaxis, []).append(t)
+ price = [t for t in fig.data if t.name == "Price" and t.x is not None and len(t.x)
+ and t.x[0] is not None]
+ oi = [t for t in fig.data if t.name == "Open Interest"]
+ assert len(price) == 1 and len(oi) == 1
+ assert price[0].yaxis == "y" and oi[0].yaxis == "y2" # OI on the secondary axis
+ hm = _heatmap(fig)
+ assert hm.yaxis == "y3"
+ assert list(hm.y) == [s.label for s in series] + ["Counterparty"]
+ assert _marks(fig) is not None and _marks(fig).yaxis == "y3"
+ assert hm.xgap == 0 # a full-width comb otherwise
+ index_lines = [t for t in _named_traces(fig) if t.yaxis == "y4"]
+ assert {t.name for t in index_lines} == {s.label for s in series}
+ assert list(fig.layout.yaxis4.range) == [0, 100]
+ # One shared time axis: every other x-axis matches the same one.
+ matches = {fig.layout[k].matches for k in fig.layout if k.startswith("xaxis")}
+ assert len(matches - {None}) == 1
+
+
+def test_flow_view_shades_exactly_the_marker_cutoffs_on_the_index_panel():
+ fig, _, _ = _flow_view()
+ rects = [s for s in fig.layout.shapes if s.type == "rect"]
+ spans = sorted((r.y0, r.y1) for r in rects)
+ assert spans == [(0, cm_const.FLOW_LEVEL_LOW), (cm_const.FLOW_LEVEL_HIGH, 100)]
+ assert all(r.yref == "y4" for r in rects)
+
+
+def test_flow_view_titles_name_the_window_and_the_triangles():
+ df = _frame(cot_categories.REPORT_DISAGG)
+ titles = ct.flow_view_titles(df)
+ assert titles[0] == "Price and Open Interest"
+ assert "52w sd" in titles[1] and "▲" in titles[1] and "▼" in titles[1]
+ assert "above 80" in titles[1] and "below 20" in titles[1]
+ assert "52w range" in titles[2]
+ assert all("—" not in t for t in titles)
+
+
+def test_flow_view_keeps_price_panel_when_the_frame_has_no_open_interest():
+ df = _frame(cot_categories.REPORT_TFF).drop(columns=["Open Interest"])
+ fig, _, _ = _flow_view(cot_categories.REPORT_TFF, df=df)
+ assert not [t for t in fig.data if t.name == "Open Interest"]
+ assert [t for t in fig.data if t.name == "Price"]
diff --git a/tests/test_flow_copy.py b/tests/test_flow_copy.py
new file mode 100644
index 0000000..309d0ec
--- /dev/null
+++ b/tests/test_flow_copy.py
@@ -0,0 +1,310 @@
+"""The week in words under the Weekly Flow panel.
+
+Store-free: every test reads the synthetic category frame from test_category_traces,
+which carries the flow family through cotmetrics' own functions.
+"""
+
+import importlib
+import sys
+
+import pandas as pd
+import pytest
+from cotmetrics import categories as cot_categories
+from cotmetrics import constants as cm_const
+from cotmetrics import flow_roles, flows
+
+import components.flow_copy as fc
+import viz_config
+import viz_constants as vc
+from tests.test_category_traces import HEADER, _frame
+
+DISAGG = cot_categories.REPORT_DISAGG
+TFF = cot_categories.REPORT_TFF
+
+
+def _by_key(report):
+ return {s.key: s for s in cot_categories.categories_for(report)}
+
+
+def _opinion(df):
+ by_key = _by_key(df.attrs["report"])
+ return [by_key[k] for k in df.attrs["flow_roles"]["opinion"]]
+
+
+def test_module_imports_without_the_data_layer():
+ for name in [m for m in sys.modules if m.startswith("cotmetrics.indexer")]:
+ del sys.modules[name]
+ importlib.reload(fc)
+ assert not any(m.startswith("cotmetrics.indexer") for m in sys.modules)
+
+
+@pytest.mark.parametrize("report", [DISAGG, TFF])
+def test_caption_opens_with_the_week_and_names_every_opinion_cohort(report):
+ df = _frame(report)
+ out = fc.week_in_words(df, HEADER)
+ date = df.index[-1]
+ assert out[0].startswith(f"Positions as of {date.strftime('%A')} "
+ f"{date.strftime('%Y-%m-%d')}")
+ for i, spec in enumerate(_opinion(df), start=1):
+ assert out[i].startswith(spec.label), out[i]
+ assert "longs " in out[i] and "shorts " in out[i]
+ assert "52-week sd" in out[i]
+
+
+def test_cohort_sentence_reports_net_legs_and_z_from_the_frame():
+ df = _frame(DISAGG)
+ spec = _opinion(df)[0]
+ i = len(df) - 1
+ df.loc[df.index[i], flows.flow_col(spec)] = -12_345
+ df.loc[df.index[i], flows.flow_long_col(spec)] = -2_000
+ df.loc[df.index[i], flows.flow_short_col(spec)] = 10_345
+ df.loc[df.index[i], flows.flow_z_col(spec)] = -1.84
+ df.loc[df.index[i], flows.flow_sign_col(spec)] = -1
+ text = fc.cohort_sentence(df, spec, i, HEADER)
+ assert text.startswith(f"{spec.label} net sold 12,345 contracts")
+ assert "(longs -2,000, shorts +10,345)" in text
+ assert "z -1.84 against its own 52-week sd" in text
+ assert "inside one sd" not in text
+
+ df.loc[df.index[i], flows.flow_z_col(spec)] = 0.4
+ df.loc[df.index[i], flows.flow_sign_col(spec)] = 0
+ assert "inside one sd" in fc.cohort_sentence(df, spec, i, HEADER)
+
+
+def test_z_at_the_threshold_prints_what_the_sign_says():
+ """1.04 is active and 0.96 is not; at one decimal both read "+1.0"."""
+ df = _frame(DISAGG)
+ spec = _opinion(df)[0]
+ i = len(df) - 1
+ for z, sign, inside in ((1.04, 1, False), (0.96, 0, True), (1.0, 0, True)):
+ df.loc[df.index[i], flows.flow_z_col(spec)] = z
+ df.loc[df.index[i], flows.flow_sign_col(spec)] = sign
+ text = fc.cohort_sentence(df, spec, i, HEADER)
+ assert f"z {z:+.2f}" in text
+ assert ("inside one sd" in text) is inside
+
+
+def test_cohort_sentence_names_a_marked_level_and_only_a_marked_one():
+ df = _frame(DISAGG)
+ spec = _opinion(df)[0]
+ i = len(df) - 1
+ mark, level = (flows.flow_level_mark_col(spec, HEADER),
+ flows.flow_from_level_col(spec, HEADER))
+ df[mark] = df[mark].astype("Int64")
+ df.loc[df.index[i], level] = 87.4
+ df.loc[df.index[i], mark] = 1
+ assert "leaving level 87.4 (above 80) of its 52-week range" in fc.cohort_sentence(
+ df, spec, i, HEADER)
+ # 19.6 is below 20 and must not print as "20".
+ df.loc[df.index[i], level] = 19.6
+ df.loc[df.index[i], mark] = -1
+ assert "leaving level 19.6 (below 20) of its 52-week range" in fc.cohort_sentence(
+ df, spec, i, HEADER)
+ df.loc[df.index[i], mark] = 0
+ assert "leaving level" not in fc.cohort_sentence(df, spec, i, HEADER)
+ # No header, no level, no error.
+ assert "leaving level" not in fc.cohort_sentence(df, spec, i, None)
+
+
+def test_masked_and_warm_up_weeks_say_why_there_is_no_reading():
+ df = _frame(DISAGG)
+ spec = _opinion(df)[0]
+ df.loc[df.index[5], flows.flow_col(spec)] = float("nan")
+ assert "no reading this week" in fc.cohort_sentence(df, spec, 5, HEADER)
+ assert "z not readable" in fc.cohort_sentence(df, spec, 6, HEADER)
+
+
+def test_thin_cohort_says_read_the_count():
+ df = _frame(DISAGG)
+ spec = _opinion(df)[0]
+ df[flows.flow_thin_col(spec)] = pd.array([True] * len(df), dtype="boolean")
+ assert "read the count" in fc.cohort_sentence(df, spec, len(df) - 1, HEADER)
+
+
+def test_counterparty_is_always_named_with_its_members_and_its_source():
+ df = _frame(DISAGG)
+ roles = df.attrs["flow_roles"]
+ members = [_by_key(DISAGG)[k].label for k in roles["counterparty"]]
+ text = fc.counterparty_sentence(df, roles, DISAGG, len(df) - 1)
+ assert text.startswith(f"On the other side, {' + '.join(members)}")
+ assert "measured" in text
+
+ roles = {**roles, "source": flow_roles.SOURCE_UNSTABLE}
+ text = fc.counterparty_sentence(df, roles, DISAGG, len(df) - 1)
+ assert "default" in text and "did not hold steady" in text
+
+ roles = {**roles, "source": flow_roles.SOURCE_DEFAULT}
+ text = fc.counterparty_sentence(df, roles, DISAGG, len(df) - 1)
+ assert "were not measured" in text and "did not hold steady" not in text
+
+
+def test_a_cohort_on_both_sides_is_said_to_be_counted_twice():
+ """Silver, copper and orange juice measure Other Reportable into the
+ counterparty while it is also an opinion cohort, so the printed figures do not
+ sum to zero; the caption has to say why."""
+ df = _frame(DISAGG)
+ roles = {**df.attrs["flow_roles"],
+ "counterparty": ("producer_merchant", "swap", "other_reportable")}
+ text = fc.counterparty_sentence(df, roles, DISAGG, len(df) - 1)
+ assert "Other Reportable is also one of the cohorts above" in text
+ assert "counted on both sides" in text
+ plain = fc.counterparty_sentence(df, df.attrs["flow_roles"], DISAGG, len(df) - 1)
+ assert "both sides" not in plain
+
+ out = fc.week_in_words(df, HEADER)
+ assert any(s.startswith("On the other side") for s in out)
+
+
+def test_neither_side_lists_the_neutral_cohorts():
+ df = _frame(DISAGG)
+ df.attrs["flow_roles"] = {**df.attrs["flow_roles"],
+ "counterparty": ("producer_merchant",),
+ "neutral": ("swap",)}
+ text = fc.neither_side_sentence(df, df.attrs["flow_roles"], DISAGG, len(df) - 1)
+ assert text.startswith("On neither side: Swap Dealers ")
+ swap = _by_key(DISAGG)["swap"]
+ df.loc[df.index[-1], flows.flow_col(swap)] = float("nan")
+ text = fc.neither_side_sentence(df, df.attrs["flow_roles"], DISAGG, len(df) - 1)
+ assert "Swap Dealers no reading this week" in text and "n/a" not in text
+ # None when every cohort is either opinion or counterparty.
+ roles = {**df.attrs["flow_roles"], "counterparty": ("producer_merchant", "swap"),
+ "neutral": ()}
+ assert fc.neither_side_sentence(df, roles, DISAGG, len(df) - 1) is None
+
+
+def test_sum_to_zero_line_only_when_every_cohort_is_present():
+ df = _frame(DISAGG)
+ assert fc.sum_to_zero_sentence(df, DISAGG)
+ missing = df.drop(columns=[flows.flow_col(_by_key(DISAGG)["swap"])])
+ assert fc.sum_to_zero_sentence(missing, DISAGG) is None
+
+
+@pytest.mark.parametrize("state,expect", [
+ ("VALUE_ACCUM", "the opinion cohorts split"),
+ ("BROAD_ACCUM", "the opinion cohorts moved together"),
+ (flows.FLOW_STATE_QUIET, "no opinion cohort moved beyond one sd"),
+ (flows.FLOW_STATE_PARTIAL, "opinion cohorts moved beyond one sd"),
+])
+def test_state_sentence_is_vocabulary(state, expect):
+ df = _frame(DISAGG)
+ i = len(df) - 1
+ df[cm_const.FLOW_STATE] = df[cm_const.FLOW_STATE].astype(object)
+ df.loc[df.index[i], cm_const.FLOW_STATE] = state
+ text = fc.state_sentence(df, df.attrs["flow_roles"], DISAGG, i)
+ assert text.startswith(f"State: {state}")
+ assert expect in text
+ assert fc.VOCABULARY_NOTE in text
+
+
+def test_state_sentence_warm_up_and_ineligible():
+ df = _frame(DISAGG)
+ assert "not readable yet" in fc.state_sentence(df, df.attrs["flow_roles"],
+ DISAGG, 0)
+ roles = {**df.attrs["flow_roles"], "state_eligible": False}
+ text = fc.state_sentence(df, roles, DISAGG, len(df) - 1)
+ assert text.startswith("No state is named on this market")
+
+
+def test_latest_row_skips_trailing_weeks_with_no_flow_and_says_so():
+ df = _frame(DISAGG)
+ for spec in _opinion(df):
+ df.loc[df.index[-2:], flows.flow_col(spec)] = float("nan")
+ assert fc.latest_row(df) == len(df) - 3
+ out = fc.week_in_words(df, HEADER)
+ assert out[1].startswith("The newest report,")
+ assert df.index[-3].strftime("%Y-%m-%d") in out[0]
+
+
+def test_caption_ends_with_the_marker_key_and_survives_no_header():
+ df = _frame(DISAGG)
+ out = fc.week_in_words(df, HEADER)
+ assert out[-1].startswith("A triangle marks")
+ assert "above 80" in out[-1] and "below 20" in out[-1]
+ assert "52-week range" in out[-1]
+ bare = fc.week_in_words(df, None)
+ assert bare and not bare[-1].startswith("A triangle")
+
+
+def test_frames_without_flow_roles_give_no_caption():
+ df = _frame(DISAGG)
+ df.attrs.pop("flow_roles")
+ assert fc.week_in_words(df, HEADER) == []
+ assert fc.week_in_words(pd.DataFrame(), HEADER) == []
+
+
+BANNED = ("mover", "biggest move", "unusual", "next", "follow", "expect",
+ "forecast", "predict", "bullish", "bearish", "likely", "edge",
+ "outperform", "return")
+
+
+@pytest.mark.parametrize("report", [DISAGG, TFF])
+def test_copy_never_ranks_or_forecasts_on_any_week(report):
+ """Every week of the frame, including the warm-up and the marked weeks."""
+ df = _frame(report)
+ for i in range(len(df)):
+ for sentence in fc.week_in_words(df, HEADER, i=i):
+ low = sentence.lower()
+ assert not any(b in low for b in BANNED), sentence
+ assert "signal" not in low.replace("not a signal", ""), sentence
+ assert "\u2014" not in sentence, sentence
+
+
+def test_caption_renders_under_the_graph_only_with_the_flow_panel(monkeypatch):
+ """The page adds markup only: the sentences are flow_copy's, in order."""
+ import dash
+ from dash import html
+
+ dash.Dash(__name__, use_pages=True, pages_folder="")
+ import pages.analytics.categories as page
+
+ df = _frame(DISAGG)
+
+ class _Indexer:
+ def get_category_data(self, asset, report, lookback):
+ return df
+
+ monkeypatch.setattr(page, "get_indexer", lambda: _Indexer())
+ keys = [s.key for s in cot_categories.categories_for(DISAGG)]
+ palette = sorted(viz_config.get_palette_names())[0]
+
+ out = page.render_category_stack(palette, "Gold", DISAGG, keys, ["net_pos", "flow"],
+ "52", "1", vc.LAYOUT_FACET)
+ assert isinstance(out, list) and len(out) == 2
+ caption = out[1]
+ assert isinstance(caption, html.Div) and caption.id == "categories_flow_caption"
+ assert [p.children for p in caption.children] == fc.week_in_words(df, HEADER)
+
+ out = page.render_category_stack(palette, "Gold", DISAGG, keys, ["net_pos"],
+ "52", "1", vc.LAYOUT_FACET)
+ assert not isinstance(out, list)
+
+
+def test_flow_view_renders_three_panels_with_the_caption_and_locks_its_controls(
+ monkeypatch):
+ import dash
+ from dash import html
+
+ dash.Dash(__name__, use_pages=True, pages_folder="")
+ import pages.analytics.categories as page
+
+ df = _frame(DISAGG)
+
+ class _Indexer:
+ def get_category_data(self, asset, report, lookback):
+ return df
+
+ monkeypatch.setattr(page, "get_indexer", lambda: _Indexer())
+ keys = [s.key for s in cot_categories.categories_for(DISAGG)]
+ palette = sorted(viz_config.get_palette_names())[0]
+ # The plot selector is ignored: net_pos alone still gives the flow view.
+ out = page.render_category_stack(palette, "Gold", DISAGG, keys, ["net_pos"],
+ "52", "1", vc.LAYOUT_FLOW)
+ graph, caption = out
+ titles = [a.text for a in graph.figure.layout.annotations]
+ assert titles[0] == "Price and Open Interest"
+ assert titles[1].startswith("Weekly Flow z")
+ assert titles[2].startswith("Positioning Index")
+ assert isinstance(caption, html.Div) and caption.children
+
+ assert page.lock_controls_for_flow_view(vc.LAYOUT_FLOW) == (True, True)
+ assert page.lock_controls_for_flow_view(vc.LAYOUT_FACET) == (False, False)