* fix(quickadapter): pin the real fractal-dimension behaviour and de-duplicate the natr defaults
Two findings from the post-merge review.
`_fractal_dimension` was reported degenerate, and the report is right:
`HL1` and `HL2` each span a HALF of the window while `HL3` spans all of it,
so `HL1 + HL2 <= 2 * HL3`, `D <= 1`, and the clip returns exactly 1.0 on every
reachable input — measured, max 0.99994 over 300000 windows. `alpha` is
therefore identically `exp(-4.6 * 0) == 1.0` and `frama` reproduces `close`
after its seed.
I first rewrote it to Ehlers' sub-range formulation. That was wrong twice
over: `range(0, period, period // 2)` with period 16 still yields two
segments, so the dimension stayed 1.0, and the new code raised on any window
shorter than `period`. Both reverted. Reconstructing that recursion from memory
is not something to ship in a coverage pass.
What lands instead is the honest form. The docstring records the invariant and
the measurement, names the paper formulation that is deliberately NOT adopted,
and the tests now ASSERT the real behaviour: `frama` reproduces `close`, and
the dimension is exactly 1.0 on trending, noisy and plateaued windows alike. The
degeneracy becomes a pinned fact rather than a silent surprise — a change that
made the dimension vary would have to update a test instead of passing unnoticed.
The period guard now also refuses values below 2, which it previously did not.
The second finding: `natr_period` and `natr_multiplier` were written down three
times with no canonical source, which `.github/copilot-instructions.md` forbids
outright. All three now derive from `DEFAULT_LABEL_NATR_PERIOD` and the
existing `DEFAULT_MIN_LABEL_NATR_MULTIPLIER`. Values are unchanged (14, 9.0), so
no behaviour moves. Verified: mutating either canonical constant now turns the
suite red, where before all four spellings drifted together unnoticed.
The test added in `
166d63c1` claimed a fixture "can see any of the three" of
the alpha constants. That was false — it is removed, replaced by the two
assertions above.
* test(quickadapter): make two data-split assertions discriminate
Both findings come from the post-merge review and both are now load-bearing.
The weight-recomposition test asserted only the lengths and mean == 1, so any
uniform vector passed, even though the production comment immediately above
names "silently uniformize" as the hazard it exists to prevent. It now asserts
the RATIOS: base is a strict ramp and the label weights are None, so a
recomposition is a pure rescale and the weights must reproduce the base
ratios, which np.ones cannot do. Verified: replacing the validation-split
composition with np.ones now fails with "Mismatched elements: 5 / 6".
The NaN-skip test could not discriminate the guard it names, because its
fixture keeps a clean second column whose value coincides either way. A new
case uses the only emitted column carrying a NaN, where the caller must get
None and fall back to the position-based purge; without the guard the NaN
reaches astype(np.int64) and raises IntCastingNaNError. Verified killed.
One caveat on how the first was verified: the repo holds several
_compose_train_weights_with_support calls at the same indentation, so the
first mutation attempt replaced the wrong call site and appeared to survive.
It only died once the mutant was anchored on the full statement.
* test(quickadapter): bound the inner validation count
The `test_size >= n_train_rows` guard was never reached: relaxing it to `>`
left all 967 tests green. It is the refusal an operator sees when the inner
validation tail is as large as the training rows remaining after the outer
holdout, so losing it turns a clear DependencyException into a numpy error
far from the cause.
The new case exercises the boundary itself — exactly eight rows, and one more —
because `>=` versus `>` is only distinguishable there. Verified killed.
* test(quickadapter): close two ways the coverage gate could be widened silently
The suite contract derived its production module list from two FLAT globs while
the measured tree is `quickadapter/user_data` recursively. The file comment
claims deriving from the tree "keeps this contract true when a source file is
added", which was false for any file outside those two directories: it reached
the coverage denominator and was subject to no contract at all. It now walks
the tree, and derives each module name from its own location — bare name under
`strategies`, dotted path anywhere else. Verified: a flat module added at
`user_data/helpers_probe.py` that fails on import now errors the whole
collection, where before nothing looked at it.
The floor test banned `omit` and nothing else. `omit` is only the first way to
shrink the denominator: `exclude_lines`, `exclude_also` and the partial_*
family are the same lever under other names. Measured on coverage 7.16.2,
exclude_lines, exclude_also and partial_branches are all [report]-only — it
REJECTS them under [run] — so the guard now checks the full set in the section
that honours them, and the contract test checks them in both. Verified:
adding `exclude_also = ^\s*raise\b` to [report] turns the floor test red.
* test(quickadapter): make the fallback warning carry its own diagnosis
Under support_policy=fallback there is no exception for the operator to read, so
the WARNING is the entire diagnostic — but the test only used assertLogs, which
proves THAT something was logged, not WHAT. Dropping a reason from the sparse
support gate still left the suite green on this path.
It now asserts the single record names the threshold that actually failed and not
one that merely happened to trip alongside it. Verified: removing the
effective_sample_size reason turns three tests red, this one included.
Also refreshes the recorded coverage measurement to 69.3 percent (floor 67, gate
exit 0).
* fix(quickadapter): report zigzag pivots as positions, not index labels
`ZigzagResult.indices` was documented at Utils.py:725 as "positions of detected
pivots in series" and consumed that way by `compute_label_weights`, which casts to
int and bounds the result by the row count. It was produced as `df.index.tolist()`
— index LABELS. The two coincide only on a zero-based RangeIndex, which is what
every fixture in this suite used, so the mismatch was invisible.
On what FreqAI actually passes it is not cosmetic. Reproduced in the QA image:
RangeIndex(0..n) -> OK, 2 non-zero weights
RangeIndex(500..500+n) -> 5/5 pivots dropped, every weight silently 0.0
DatetimeIndex -> TypeError: int() argument ... not 'Timestamp'
So any config with `freqai.label_weighting.strategy` other than "none" crashes
`set_freqai_targets` on the first candle of every pair under live and backtest,
and degrades to zero label weighting — with only a warning — on any offset
integer index. The bug predates this suite; the suite is what hid it.
`_zigzag` now appends `pos`, the series assignment uses `.iloc`, and both sites
carry the reason. All three index shapes now agree exactly.
The suite could not have caught this on its own, so two tests were added that can:
one asserts the pivot positions and the label values are invariant across a
shifted, a datetime and a string index; the other drives `compute_label_weights`
end to end and requires a non-zero weight at every pivot. One existing assertion
read `label.series.loc[position]`, which only worked because labels and positions
coincided, and now uses `.iloc`. Verified: reintroducing `df.index[pos]` fails
five tests.
* fix(quickadapter): correct the frama degeneracy claim and pin the period guard
The `_fractal_dimension` docstring asserted two things that are false, and I wrote
them. It claimed "the clip therefore returns exactly 1.0 on every path, so alpha
is identically exp(-4.6 * 0) == 1.0 and frama reproduces close".
The inequality holds — HL1 and HL2 each span a half window while HL3 spans all of
it, so D <= 1 — but the conclusion does not. `np.clip(D, 1.0, 2.0)` is a no-op
for a value already at or above the lower bound, so it passes a ratio of
1.
0000000000000002 straight through. Measured on a deterministic ten-bar window:
D = 1.
0000000000000002, alpha = 0.
999999999999999. `frama` reproduces `close` to
within a few ULP, not bit-exactly, and a short-circuit on `alpha == 1.0` would
never fire.
The docstring now says what is actually true, states the deterministic
counterexample, and says explicitly that nothing downstream may treat alpha as a
literal 1.0. The test named `..._is_exactly_one_on_every_window` asserted
`== 1.0` on three hand-picked windows that happen to land exactly on 1.0, so it
was pinning a rounding accident rather than the contract; it now asserts within
rounding, and a new test carries the counterexample that makes it honest.
Separately, the `period < 2` clause of the guard had no test. The existing case
only drives odd periods, which the evenness clause already catches, so the clause
could be dropped with a green suite — and dropping it hands the caller numpy's
"zero-size array to reduction operation maximum which has no identity" instead of
the documented domain error. Zero is even, so it is the only value that exercises
the clause on its own; that is now the case.
One note on the mutant: rewriting the clause to `period < 1` leaves the suite
green, and that mutant is equivalent, not surviving — period 1 is odd and already
refused. The clause is only killable by removing it, which turns two tests red.
* test(quickadapter): close the four remaining ways the coverage gate goes inert
The gate was declared trustworthy and none of it was enforced. Each of the four
below was verified by mutation, and each now turns the suite red.
A `# pragma: no cover` in a production file. coverage.py always applies a
default `exclude_list` matching that pattern, so it is the source-side twin of
the `omit` key guarded two tests above, and the source side was open. Measured:
one pragma on `_normalize_final_take_profit_state` moved QuickAdapterV3.py from
58.8 % to 60.2 % and the total from 69.3 % to 69.6 %, with every test green. At
2.3 points of headroom the move is repeatable at every uncovered `def`.
The `fail_under` value. `.coveragerc` and README.md:528 both say "raise only,
never lower"; `fail_under = 5` passed everything. The floor test now pins the
lower bound and leaves raising it free, which is the direction the policy allows.
The provenance annotation. It had to contain the word "measured"; it did not have
to contain a date, and `# measured whenever someone remembered` passed. It now
has to record a well-formed date and a real percentage, and that percentage has
to clear the floor it justifies — a floor of 70 annotated with a 69.3 %
measurement is a gate that cannot pass.
Whether the CI step propagates the result. `exit $status` is the only thing
turning a failing `coverage report` into a failing job, no test read the
workflow, and deleting that one line left the whole class green. Verified: with
`fail_under = 99` the step exits 2 with the line and 0 without it.
One residual, deliberate: an old but well-formed date such as 1970-01-01 still
passes. Refusing a date older than N months would make the suite fail on a
repository nobody has touched, which is a time bomb rather than a gate. The
annotation has to be checkable; whether it is current stays a reviewer's call.
* test(quickadapter): pin the volume-weighted efficiency ratio and realign the README
`volume_weighted_efficiency_ratio` is the sixth COMBINED_METRIC feeding label
importance, and it was the only one with no numeric assertion: every check either
mirrored `efficiency_ratio` or expected 1.0, which holds for ANY positive
weighting. The W_LEGS fixture is monotone, so its ratio is 1.0 under every
convention and cannot discriminate. Shifting the volume window by one candle —
same length, different slice — left all 968 tests green.
The new case uses a slow uptrend carrying a fast oscillation, so the legs are
short and genuinely non-monotone and the ratio drops well below 1.0. It
restates the documented convention as an oracle over every leg: each
close-to-close step is weighted by the volume of the candle it lands on, and the
pivot bar at the start of the leg is excluded from the total. It then asserts
that the one-candle shift a mutation introduces gives a different answer, so the
convention is load-bearing rather than an identity. Verified: the window shift
now turns three tests and one error red.
Worth recording, because it cost time: I first tried six explicit shapes for the
fixture and every one gave exactly 1.0, which looked like proof the metric was
constant and therefore dead. It is not — across 3401 legs of random walks, 2941
differ from 1.0 with a wide spread. Zigzag legs are usually monotone, so a
sampled fixture is the only way to reach the non-trivial branch.
The README paragraph the previous commit edited was 99 columns, the widest prose
line in a 190 KB file whose prose wraps at 79; the widest before was 84. Rewrapped.
The coverage-gate paragraph also claimed the floor test refuses "any omit", which
stopped being true two commits ago, and understated what is now enforced.
* test(quickadapter): name a fractal-dimension test after what it asserts, align docstrings
`test_a_framed_gap_stops_the_fractal_dimension_at_the_window_edge` was named for
a gap and a window edge its fixture cannot contain — it feeds a constant frame, so
the only branch it reaches is the zero half-range short-circuit. The name read as
if the early return were covered from a second direction, which it is not.
Renamed to the assertion it actually makes: a constant window yields a FINITE
fractal dimension rather than log(0) and a -inf alpha. That guard is real and
distinct from the neighbouring constant-series case, so the test stays; the
overlapping tail assertion went, and a NaN check was added to match the name.
Twenty of twenty-eight test modules carried the "; requires the Freqtrade QA
image." clause and eight did not, which is the kind of drift the repo rules call
incoherent. All twenty-eight now carry it.
Left alone on purpose: `test_prediction_history.py` and
`test_regressor_prediction_history.py` are near-synonym names, and the new one is
the better of the two — the pre-existing module is what reads as the odd one out.
Renaming it would break a stable public path for a naming preference.
* test(quickadapter): guard the positional pivot contract at the strategy level
The previous commit proved the fix by hand — driving the real strategy path with
a DatetimeIndex frame and with a shifted integer index, both of which now give
184 labels and 23 non-zero extrema weights. That evidence was not kept, and the
suite still could not catch a regression: every fixture in the suite builds `date`
as a COLUMN and leaves a zero-based RangeIndex, which is exactly the shape that
hides the bug.
This runs `set_freqai_targets` on the real strategy with the index set both ways
and requires identical labels and identical weights, plus the absolute counts so
that a column of zeros or an all-positive column cannot satisfy it. Note the
weight count is 195, not 23: 23 of those sit on pivots and the rest are the
non-causal baseline fill.
Verified: restoring `.loc` at Utils.py:846 turns eight tests red, and appending
`df.index[pos]` turns five red.
* test(quickadapter): close the gaps the second review tour found
Every item below was confirmed by mutation before and after. Three were defects in
the previous tour's own work, and are called out as such.
The production default had no coverage at all. `_make_train_test_split_datasets`
is what a stock configuration runs — `causal_mode` defaults to True, the split
method defaults to `train_test_split`, `config-template.json` ships
`"test_size": 0.333` — and no test called it. Every split test drove
`_add_validation_split` or `test_size: 0` instead, so the whole causal purge block
ran zero times and both of its boundary comparisons were unguarded: relaxing `<`
to `<=` on the label horizon, or on the label's known_at position, kept the suite
green. Four cases now drive the outer split and pin each boundary from both sides,
including the counterpart with `causal_mode` off so they cannot pass for the wrong
reason. Both off-by-one mutants are killed.
The monotonic pivot watermark had an untested term. `add_pivot` takes the max of
four, and dropping `confirmed_at_pos` left the suite green while publishing labels
before their confirmation candle on 41 of 1500 seeded frames, by up to 96 candles.
The existing case cannot tell the term from `natr_warmup_end_pos` because on its
fixture the two coincide. A flat stretch followed by a trend separates them.
`compose_label_lookahead` had no test whatsoever. Its right-edge clamp was a row
wide in the wrong direction without anything noticing.
Three of these are mine. `test_the_workflow_propagates_the_coverage_result` scoped
its `exit $status` search on an unindented `- name:`, which never occurs in the
file, so it ran to end-of-file and a decoy token anywhere later satisfied it; it
now scopes on the step's own indent. It also never checked that the workflow points
at `.coveragerc` at all, so deleting one `--env` line left coverage with no
configuration, measuring the standard library and exiting 0. And the pragma guard's
comment claimed the tree was free of coverage's default exclusions when
`if TYPE_CHECKING:` is present in Utils.py; the guard now counts all three
DEFAULT_EXCLUDE patterns and pins the legitimate one.
Also: `include` was missing from the exclusion-key list, so one `[report] include`
line cut the denominator from 6217 statements to 2717 silently. `precision` was
checked for presence but never for value, and coverage's predicate is
`round(total, precision) < fail_under`, so `precision = 0` let a run half a point
below the floor pass.
And four tests asserted nothing: a tautological `assertLess` over a slice that
made it true by construction, now a behavioural test that the renamed section
reaches its keys through the second hop; a round-trip-only flat-config test that
any invertible pipeline satisfies, now asserting the resolved standardization and
normalization; a memoisation test that could not see a leak because each probe
rotates the timestamp and wipes the cache, now asserting after every refusal; and
`assertIsNotNone(import_module(...))`, which never returns None, deleted.
* docs(quickadapter): correct the coverage-gate citations and complete the guard list
Two comments pointed at README.md:528 for the "raise only, never lower" policy,
which is on line 532; 528 is the start of the surrounding paragraph. A line-number
citation into a file the same change edits is worth getting right.
The README paragraph listing what the floor test refuses named nine of the guards
and omitted three, one of which matters: it never said that `fail_under` drifting
into the inert `[run]` section is refused, so a maintainer reading it could "fix" a
red build by moving the line and believe the gate still holds. The list now covers
all sixteen, including the ones added since it was written.
Recorded coverage moves to 69.6 percent (floor 67, gate exit 0), from the same
configuration as the 69.3 percent it replaces.
* docs(quickadapter): correct the documented failure mode of an inline env prefix
README.md:518 said that prefixing `COVERAGE_RCFILE` and `COVERAGE_FILE` on a
single command makes `coverage report` "read no configuration and exit 0 while
measuring nothing". Reproduced in the QA image on a clean checkout: it prints
`No data to report.` and exits 1. The second command looks for the data file in
its default location, finds none, and fails.
The advice to export both variables is right, and the failure is loud rather than
silent — it cannot ship a green gate on a measurement that never happened — so the
text now says that instead. What it cannot do is send someone looking for a silent
zero exit that does not happen.
* test(quickadapter): pin how the two causal masks combine, and the flat-candle refusal
Both survivors were in tests added two commits ago, found by the third tour.
The causal purge applies two masks and the first version of that class proved
each one separately while never proving they combine: the horizon case carried no
availability column, the availability case used a zero horizon, so `keep_mask &=`
never ran with both constraints live. Rewriting the conjunction as an assignment
discards the horizon mask and kept the suite green. The new case exercises both
directions, because either alone is satisfied by a mask that simply overwrites the
other — the horizon removing rows the availability mask keeps, and the availability
mask removing a row the horizon kept.
I first wrote that case backwards, choosing a row the horizon mask keeps and
asserting it was purged. The reviewer caught it before it was committed. The
lesson is recorded in the test: pinning constraints one at a time says nothing
about how they combine.
The flat-candle threshold refusal was never checked against `_candle_threshold_cache`.
Every later probe rotates the timestamp on purpose, so a refusal that memoised was
wiped by the next invalidation and unobservable at the end of the test. It is now
asserted while that frame's signature is still live. Verified: inserting a memo in
that branch — with either `0.0` or `np.nan` — turns two subtests red, where before
both survived.
* test(quickadapter): reach the validation modes and weighting metrics that no test did
A block that never executes is worse than a weak assertion, because it reads as
coverage. The non-finite scalar case put the warn and none modes inside the same
`with` block as the raising call, so the raise on the first line unwound the block
and neither branch ever ran. The test name claimed all three modes. Split into
three, and the two mutants the reviewer measured — removing the non-finite
warning, and turning its `return None` into `return 0.0` — now both fail.
The per-metric weighting strategies had no test at all. Only `uniform` was ever
selected, so `weights = np.asarray(metrics[strategy])` never ran and replacing it
with `np.ones` left the suite green while every configured swing importance became
uniform. The amplitude strategy is reachable from a real config and is the
alternative the shipped template documents. The new cases pin the metric on the
pivots and the ratio the training-sample-weight consumer actually sees, since
the mean and positivity hold either way.
The reversal decay test asserted only that the decayed maximum is at least the
decayed minimum. That ordering follows from the earlier rejection of `lo > hi`,
so it holds for any input and says nothing about the decay; dropping the decay on
the maximum, or squaring the factor, both survived. The historical maximum is now
pinned at [1.0, 0.5, 0.25].
The eval weight test asserted a unit mean, which a binarisation of the label
weights also satisfies. It now pins the ratios.
And the migration table has two-hop chains that a structural "a chain exists"
assertion plus one behavioural chain did not cover: `threshold_outlier` reaching
`outlier_quantile`, `extrema_fraction` reaching `keep_fraction`, and a key under
the renamed `predictions_extrema` section reaching `threshold_method`. Sorting the
table by old path — which strands all three — now fails.
* test(quickadapter): guard the CI conditions that dispatch the gate, and a logger leak
Three passes closed the ways the coverage CONFIGURATION could be made inert. This
closes the level above it: whether the step runs at all. Flipping the QuickAdapter
matrix entry to `coverage: false`, or the step's own `if:` to a constant, left every
other test green while CI stopped running the gate — and a job that skips the
coverage step is indistinguishable from a green one. The matrix and the step
condition are now parsed rather than grepped, so an entry that loses its key, or an
`if:` that stops referencing the matrix, fails here too. All three mutants die.
`recorded_warnings` set its logger to NOTSET and never restored the level. Two
distinct problems, and only the second is fixed by restoring it. NOTSET makes the
logger inherit the root level, and a record the logger has already filtered out
never reaches a private handler: with the root at ERROR, all 45 cases in that file
failed on empty sinks. The capture level is now set explicitly, and the previous
level is restored on exit so later tests do not inherit it. Verified: 89 failures
before, 0 after.
* test(quickadapter): cover the chronological split, whose whole body never ran
`_make_timeseries_split_datasets` is reachable through the documented
`data_split_parameters.method='timeseries_split'`, and coverage reported 0 of its
62 statements executed. Its own availability comparison at
QuickAdapterRegressorV3.py:3108 could go from strict to inclusive, or be deleted,
with the suite green — a training row whose label is first knowable on the first
test row would be admitted into the fit.
It is a sibling of the block already pinned, not a subclass of it. Inheriting
looked like the economical choice and was wrong: the parent's expectations are
written around a ten-row test tail starting at position 30, where this last fold
is five rows starting at 35, and five inherited tests failed asserting the wrong
geometry. Each carries its own fixtures now, and the docstring records why.
The boundary needed measuring rather than assuming. `gap=2` removes the two rows
before the test fold from the training set BEFORE the availability comparison, so
the last row that comparison can act on is 32, not 34: measured train ends at 32
and test is 35..39. A lookahead placed on 33 or 34 is indistinguishable from no
lookahead at all. The cases now walk the last train row across availability at
32, 33 and 34 — all strictly before the cutoff — and exactly at 35, which is
purged. Verified: relaxing `<` to `<=` turns two tests red.
Coverage moves to 70.5 percent (floor 67, gate exit 0).
* test(quickadapter): pin the requested threshold method and the take-profit lower bound
Two survivors from the third tour, both in tests written before this review.
A range check cannot tell delegation from substitution. Every skimage threshold
method lands inside [-1, 1] whatever it computes, so forcing `threshold_func` to
Otsu inside the consumer left the whole suite green while `mean` silently became
Otsu's -0.9485 instead of -0.55 — an entry gate using a method the config never
asked for. The full tuple is now pinned per method. The whole table, not just the
differing entry: on this fixture `yen` coincides with `otsu` and `li` with `mean`,
so pinning a single method would leave the substitution open from the other side.
Verified: rerouting every method to Otsu, and rerouting only `yen` to `mean`, each
turn six subtests red.
The closed-interval case for the exit distances called only the stoploss helper,
so the take-profit endpoints were untested. Relaxing its valid condition to a
strict lower bound, or returning None for a zero fraction before the range check,
each left the suite green. Both helpers are now called at both endpoints. The upper
endpoint was already covered by sibling tests; the lower one is only observable
here, and the strict-lower mutant now dies.
python -m coverage report
```
-Export both variables rather than prefixing a single command: an inline
-prefix binds to that command only, so `python -m coverage report` would read
-no configuration and exit 0 while measuring nothing.
+Export both variables rather than prefixing a single command: an inline prefix
+binds to that command only, so `python -m coverage report` looks for the data
+file in its default location, finds none, and exits 1 with `No data to report.`
+The mistake is loud rather than silent — it cannot ship a green gate on a
+measurement that never happened — but it does break a run that would otherwise
+have finished.
Branch coverage is required. Every guard in this codebase is an early return
or a raise, and statement coverage marks a guard covered the instant its `if`
is not a valid input — then update the value and the `# measured` annotation
above it in the same commit. Raise the floor only; a drop needs the reason in
the pull request. `quickadapter/tests/test_coverage_floor.py` refuses a
-placeholder, a missing measurement annotation, a disabled branch trace, a
-disabled namespace walk, and any `omit` of production code.
+placeholder, a missing or undated measurement annotation, a measurement below
+the floor it justifies, a floor below the current minimum, a floor that is not a
+percentage, a `precision` coarse enough to round the total past the floor, a
+disabled branch trace, a disabled namespace walk, a source tree that is not the
+measured one, a `fail_under` that has drifted into the inert `[run]` section, the
+three of coverage's default source exclusions, a coverage pragma in production
+source, a workflow that is not pointed at this `.coveragerc`, a workflow that
+stops propagating the report status, and any `omit`, `include`, `exclude_lines`,
+`exclude_also` or `partial_*` of production code.
### Quality checks
dependency versions remain exact.
The BasedPyright and type-stub versions are pinned in each project's
-`.devcontainer/requirements-dev.txt`, as is `coverage` in QuickAdapter's. The Freqtrade base images
-intentionally follow their rolling `stable_freqai` and `stable_freqairl` tags,
-so record the resolved image digests when a reproducible audit is required.
+`.devcontainer/requirements-dev.txt`, as is `coverage` in QuickAdapter's.
+The Freqtrade base images intentionally follow their rolling `stable_freqai`
+and `stable_freqairl` tags, so record the resolved image digests when a
+reproducible audit is required.
## Common workflows
show_missing = True
precision = 1
include_namespace_packages = True
-# measured 2026-10-01 at 69.2% (23.9% before this suite, same configuration) — raise only, never lower
+# measured 2026-10-01 at 70.5% (23.9% before this suite, same configuration) — raise only, never lower
fail_under = 67
"""Guards the coverage gate itself; requires the Freqtrade QA image."""
import configparser
+import re
import unittest
-from qa_support import COVERAGERC, QaTestCase
+import yaml
+from qa_support import COVERAGERC, REPO_ROOT, QaTestCase
FLOOR = "fail_under"
PLACES = "precision"
BRANCH = "branch"
NAMESPACE_PACKAGES = "include_namespace_packages"
OMIT = "omit"
+# The lowest floor the project will accept. Raising it is free; dropping below this is the
+# silent gate disabling that `.coveragerc` and README.md:532 both forbid.
+MINIMUM_FLOOR = 67.0
+PRAGMA = re.compile(r"pragma\s*:\s*no\s*cover", re.IGNORECASE)
+# The other two of coverage.py's three DEFAULT_EXCLUDE patterns.
+ELLIPSIS_BODY = re.compile(r"^\s*(((async )?def .*?)?[\])]+(\s*->.*?)?:\s*)?\.\.\.\s*(#|$)")
+TYPE_CHECKING = re.compile(r"^\s*if (typing\.)?TYPE_CHECKING:")
+INCLUDE = "include"
+MEASURED_TREE = REPO_ROOT / "quickadapter" / "user_data"
+WORKFLOW = REPO_ROOT / ".github" / "workflows" / "quality.yml"
class CoverageFloorTest(QaTestCase):
def test_unexecuted_files_stay_in_the_denominator(self):
self.assertTrue(self.parser.getboolean("report", NAMESPACE_PACKAGES))
- def test_precision_is_set_so_the_floor_is_interpreted_consistently(self):
- self.assertIn(PLACES, self.parser["report"])
+ def test_precision_is_fine_enough_for_the_gate_to_compare_what_it_prints(self):
+ # coverage's predicate is `round(total, precision) < fail_under`, so precision is not
+ # cosmetic: at 0 the measured total is rounded to a whole number and a run up to half a
+ # point BELOW the floor passes. Measured with the shipped floor of 67, a true total of
+ # 66.6 fails at precision 1 and passes at precision 0, and the same divergence reaches
+ # `coverage report`'s exit code through the CLI. The key was present; its value was
+ # never read, so `precision = 0` shipped with the suite green.
+ precision = self.parser.getint("report", PLACES)
+ self.assertGreaterEqual(precision, 1)
def test_the_floor_is_bound_in_the_section_coverage_reads_it_from(self):
# coverage binds fail_under as `report:fail_under` ONLY. The same key under
self.assertIn(FLOOR, self.parser["report"])
self.assertNotIn(FLOOR, self.parser["run"])
- def test_no_production_code_is_omitted(self):
- # An omit entry would shrink the denominator while the floor stayed put, so the
- # gate would report the same number over a smaller measurement. Both sections are
- # checked: coverage accepts omit as run:omit as well as report:omit.
- for section in ("run", "report"):
- with self.subTest(section=section):
- self.assertNotIn(OMIT, self.parser[section])
+ def test_no_production_code_is_excluded_by_any_other_key(self):
+ # `omit` was only the first way to shrink the denominator: `exclude_lines`,
+ # `exclude_also` and the partial_* family are the same lever under other names, and
+ # adding `exclude_also = ^\s*raise\b` was measured to raise the reported total from
+ # 69.2 % to 70.1 % with every other floor test still green. Measured against coverage
+ # 7.16.2, all of these are `[report]`-only; it REJECTS exclude_lines, exclude_also
+ # and partial_branches under `[run]`, so checking them in both sections is free and
+ # forward-looking.
+ for section, keys in (
+ ("run", (OMIT, INCLUDE)),
+ (
+ "report",
+ (
+ OMIT,
+ INCLUDE,
+ "exclude_lines",
+ "exclude_also",
+ "partial_branches",
+ "partial_also",
+ "partial_branches_always",
+ ),
+ ),
+ ):
+ for key in keys:
+ with self.subTest(section=section, key=key):
+ self.assertNotIn(key, self.parser[section])
+
+ def test_the_floor_is_never_lowered(self):
+ # `.coveragerc` and README.md:532 both say "raise only, never lower", but nothing
+ # enforced it: `fail_under = 5` kept every test in this class green while the gate
+ # stopped constraining anything. The floor is a control the project believes it has,
+ # so the floor test now pins its LOWER bound. Raising it stays free, which is exactly
+ # the direction the policy allows.
+ self.assertGreaterEqual(float(self._raw_floor()), MINIMUM_FLOOR)
+
+ def test_the_provenance_records_a_date_and_a_measurement_that_clears_the_floor(self):
+ # The class comment claims the measured date is what tells a reader whether the
+ # number is current, but the original assertion only looked for the word "measured":
+ # both `# measured whenever someone remembered` and a bare date passed. Two things
+ # are actually checkable here, and both are asserted.
+ #
+ # Recency is deliberately NOT one of them. Refusing a date older than N months would
+ # make this suite fail on a repository nobody has touched, which is a time bomb, not
+ # a gate. What is enforceable is that the annotation records a real measurement and
+ # that the measurement clears the floor it justifies: a floor of 70 annotated with a
+ # 69.3% measurement is a gate that cannot pass, and the stale number is the symptom.
+ lines = [line.strip() for line in self.text.splitlines()]
+ index = next(i for i, line in enumerate(lines) if line.startswith(f"{FLOOR} ="))
+ annotation = " ".join(lines[max(0, index - 2) : index])
+ with self.subTest(requirement="a well-formed measurement date"):
+ self.assertRegex(annotation, r"\d{4}-\d{2}-\d{2}")
+ with self.subTest(requirement="a recorded measurement"):
+ found = re.search(r"measured[^%]*?([0-9]+(?:\.[0-9]+)?)%", annotation)
+ self.assertIsNotNone(found, f"no percentage recorded above {FLOOR}: {annotation!r}")
+ self.assertGreaterEqual(
+ float(found.group(1)),
+ float(self._raw_floor()),
+ "the annotated measurement must clear the floor it justifies",
+ )
+
+ def test_no_production_source_widens_coverage_s_default_exclusions(self):
+ # The source side of the same lever the config-side keys above cover. coverage.py
+ # always applies a DEFAULT_EXCLUDE of three patterns, and each removes statements from
+ # the denominator without any config change at all: a `# pragma: no cover` comment, a
+ # `...`-only body, and an `if TYPE_CHECKING:` block. Measured for the pragma: one on
+ # `_normalize_final_take_profit_state` took QuickAdapterV3.py from 58.8 % to 60.2 % and
+ # the total from 69.3 % to 69.6 %, with the suite green.
+ #
+ # The tree is NOT free of the lever today: `if TYPE_CHECKING:` appears in Utils.py and
+ # is excluded today. A type-checking block is legitimate — it holds imports that do not
+ # run — so it is permitted up to a pinned count rather than banned outright. The other
+ # two patterns have no legitimate use here, so they must stay at zero.
+ counts = {
+ "pragma: no cover": 0,
+ "ellipsis body": 0,
+ "if TYPE_CHECKING:": 1,
+ }
+ found = dict.fromkeys(counts, 0)
+ for path in sorted(MEASURED_TREE.rglob("*.py")):
+ for line in path.read_text().splitlines():
+ if PRAGMA.search(line):
+ found["pragma: no cover"] += 1
+ elif ELLIPSIS_BODY.match(line):
+ found["ellipsis body"] += 1
+ elif TYPE_CHECKING.match(line):
+ found["if TYPE_CHECKING:"] += 1
+ self.assertEqual(counts, found, "these shrink the denominator without any config change")
+
+ def test_the_workflow_runs_the_gate_with_this_configuration_and_propagates_it(self):
+ # Everything above constrains the CONFIGURATION to be meaningful; these two constrain
+ # something to actually read it and to act on the result.
+ #
+ # The rcfile binding: coverage looks for `.coveragerc` in the directory it runs from
+ # and does not search parents, and the QA image has no `.coveragerc`, `setup.cfg`,
+ # `tox.ini` or `pyproject.toml` at the workspace root. Deleting the one
+ # `--env COVERAGE_RCFILE=` line therefore leaves coverage with NO configuration: branch
+ # tracing off, no `source`, `fail_under = 0`, measuring the standard library and
+ # exiting 0. Nothing else in this class is sensitive to that line.
+ workflow = WORKFLOW.read_text()
+ self.assertRegex(
+ workflow,
+ r"--env COVERAGE_RCFILE=\S*\$\{\{ matrix\.context \}\}/\.coveragerc",
+ "the coverage step must be pointed at the context's .coveragerc",
+ )
+ # The propagation: `exit $status` is the only thing that turns a failing
+ # `coverage report` into a failing job. It is scoped to the COVERAGE STEP, matched on
+ # its own indent: every `- name:` in the file is indented, so searching for an
+ # unindented one silently runs off the end of the file and a decoy token anywhere
+ # later in the workflow satisfies the assertion.
+ step = re.search(
+ r"^ - name: Run runtime regressions with coverage$.*?(?=^ - name:|\n\S)",
+ workflow,
+ re.DOTALL | re.MULTILINE,
+ )
+ self.assertIsNotNone(step, "the coverage step is gone from the workflow")
+ self.assertIn("python -m coverage report", step.group(0))
+ self.assertIn("exit $status", step.group(0), "the step must exit with the report status")
+
+ def test_the_workflow_actually_dispatches_the_coverage_step(self):
+ # The previous guards constrain the CONFIGURATION and the body of the coverage step.
+ # This constrains whether the step runs AT ALL, which is the same lever one level up:
+ # flipping the QuickAdapter matrix entry to `coverage: false`, or the step's own `if:`
+ # to a constant, leaves every other test green while CI stops running the gate
+ # entirely. A job that skips the coverage step is indistinguishable from a green one.
+ #
+ # Parsed rather than grepped: a matrix entry that loses its `coverage` key, or an `if:`
+ # that stops referencing the matrix, must fail here rather than pass a substring test.
+ workflow = yaml.safe_load(WORKFLOW.read_text())
+ entries = workflow["jobs"]["strategy-qa"]["strategy"]["matrix"]["include"]
+ by_name = {entry["name"]: entry for entry in entries}
+
+ self.assertIn("QuickAdapter", by_name)
+ self.assertIs(
+ by_name["QuickAdapter"].get("coverage"),
+ True,
+ "the QuickAdapter matrix entry must request the coverage run",
+ )
+ steps = workflow["jobs"]["strategy-qa"]["steps"]
+ coverage_steps = [
+ step for step in steps if step.get("name") == "Run runtime regressions with coverage"
+ ]
+ self.assertEqual(1, len(coverage_steps), "the coverage step must exist exactly once")
+ self.assertEqual(
+ "matrix.coverage",
+ coverage_steps[0].get("if"),
+ "the coverage step must be dispatched by the matrix flag, not a constant",
+ )
def test_the_source_is_the_measured_tree(self):
self.assertEqual(
-"""Label scaling pipeline contract: round trip, non-finite handling and fit-time refusal."""
+"""Label scaling pipeline contract: round trip, non-finite handling and fit-time refusal.; requires the Freqtrade QA image."""
import unittest
def test_a_flat_config_dict_is_read_as_the_default_block(self):
# The regressor hands over whatever get_label_pipeline_config produced, which may carry
# pipeline keys at the top level instead of under "default".
+ #
+ # A round trip alone is satisfied by ANY invertible pipeline, including one that
+ # ignores the flat dict entirely and falls back to `DEFAULTS_LABEL_PIPELINE`
+ # (standardization "none", normalization "maxabs"). Two mutations proved it: dropping
+ # the config merge and dropping the key filter each left the suite green while the
+ # fitted scaler changed. So this asserts what the config bought, and that the
+ # unknown key was dropped rather than carried.
matrix = label_matrix()
fitted = LabelTransformer(
label_transformer={"standardization": "zscore", "normalization": "minmax", "ignored": 1}
)
fitted.fit(matrix)
scaled, *_ = fitted.transform(matrix)
+
+ self.assertEqual(fitted._config.default["standardization"], "zscore")
+ self.assertEqual(fitted._config.default["normalization"], "minmax")
+ self.assertNotIn("ignored", fitted._config.default)
+ # minmax pins the extremes, and that is what the `DEFAULTS_LABEL_PIPELINE` fallback
+ # (none/maxabs) cannot produce: maxabs maps into (0, 1], so its minimum is strictly
+ # positive. The column's MEAN is deliberately not asserted — minmax is affine and runs
+ # after zscore, so centring does not survive it.
+ column = scaled[:, 0]
+ self.assertAlmostEqual(float(column.min()), -1.0, places=12)
+ self.assertAlmostEqual(float(column.max()), 1.0, places=12)
recovered, *_ = fitted.inverse_transform(scaled)
np.testing.assert_allclose(recovered, matrix, rtol=ROUND_TRIP_RTOL, atol=ROUND_TRIP_ATOL)
-"""Runtime contracts for FreqAI's native QuickAdapter prediction history."""
+"""Runtime contracts for FreqAI's native QuickAdapter prediction history.; requires the Freqtrade QA image."""
import unittest
from types import SimpleNamespace
import numpy as np
import pandas as pd
+from freqtrade.exceptions import DependencyException
from qa_support import QaTestCase, model_config, temporary_directory
from Utils import (
LABEL_COLUMNS,
result = QuickAdapterRegressorV3._known_at_lookahead(unfiltered, unfiltered)
np.testing.assert_array_equal(result.to_numpy(), np.array([9, 9]))
+ def test_a_lone_column_holding_a_nan_leaves_no_usable_lookahead(self):
+ # The skip is per COLUMN, and the fixture above keeps a clean weight column, so the
+ # surviving values coincide whether or not the guard runs. It is observable only
+ # when the ONLY emitted column carries a NaN: the caller must then get None and fall
+ # back to the position-based purge, where without the guard the NaN reaches
+ # astype(np.int64) and raises IntCastingNaNError out of the function.
+ unfiltered = frame(pd.Index([0, 1]), **{LOOKAHEAD: [3.0, np.nan]})
+ self.assertIsNone(QuickAdapterRegressorV3._known_at_lookahead(unfiltered, unfiltered))
+
def test_result_is_aligned_to_the_filtered_index(self):
unfiltered = frame(pd.Index([0, 1, 2]), **{LOOKAHEAD: [5, 6, 7]})
result = QuickAdapterRegressorV3._known_at_lookahead(frame(pd.Index([2, 0])), unfiltered)
result = self._split(model)
self.assertTrue(result["validation_features"].empty)
+ def test_an_oversized_validation_count_is_refused(self):
+ # test_size is bounded by the rows remaining after the outer holdout, and the bound
+ # is `>=`: eight training rows admit at most a seven-row validation tail. Removing the
+ # guard, or relaxing it to `>`, previously left the suite green.
+ for test_size in (8, 9):
+ with self.subTest(test_size=test_size):
+ model = self._model(
+ data_split_parameters={"test_size": test_size, "shuffle": False}
+ )
+ with self.assertRaisesRegex(DependencyException, "is not smaller than the 8"):
+ self._split(model)
+
def test_the_chronological_tail_is_reserved_not_sampled(self):
model = self._model(data_split_parameters={"test_size": 2, "shuffle": False})
result = self._split(model)
result = self._split(model)
self.assertEqual(len(result["train_weights"]), len(result["train_features"]))
self.assertEqual(len(result["validation_weights"]), len(result["validation_features"]))
- # A recomposition renormalises each subset to mean 1, so the recomposed train weights
- # cannot equal the raw prefix of the base weights the caller passed in.
- self.assertAlmostEqual(float(np.mean(result["train_weights"])), 1.0)
- self.assertAlmostEqual(float(np.mean(result["validation_weights"])), 1.0)
+ # The ratios are the discriminating assertion: `base` is a strict ramp and the label
+ # weights are None, so a recomposition is a pure rescale and the weights reproduce
+ # the BASE ratios. np.ones has ratio 1.0 everywhere and cannot pass this, which is
+ # the whole point of the contract the production comment spells out.
+ for weights, expected in (
+ (result["train_weights"], [1.0, 2.0, 3.0, 4.0, 5.0, 6.0]),
+ (result["validation_weights"], [1.0, 8.0 / 7.0]),
+ ):
+ with self.subTest(weights=list(weights)):
+ self.assertAlmostEqual(float(np.mean(weights)), 1.0)
+ np.testing.assert_allclose(
+ np.asarray(weights) / weights[0], np.array(expected), rtol=1e-12, atol=0.0
+ )
def test_holdout_rows_available_before_the_window_end_are_kept(self):
model = self._model(data_split_parameters={"test_size": 2, "shuffle": False})
)
+class TrainTestSplitCausalPurgeTest(QaTestCase):
+ """The OUTER split, which is the shipped default.
+
+ `causal_mode` defaults to True, the split method defaults to `train_test_split`
+ and `config-template.json` ships `"test_size": 0.333`, so
+ `_make_train_test_split_datasets` with a non-zero test size is the path a stock
+ configuration takes. Every other test drives `_add_validation_split` or `test_size: 0`
+ instead, which left the whole causal purge block — and both of its boundary
+ comparisons — with no coverage at all: relaxing `<` to `<=` on either of them kept the
+ suite green.
+ """
+
+ ROWS = 40
+ TEST_ROWS = 10
+
+ class _Kitchen:
+ pair = PAIR
+
+ @staticmethod
+ def build_data_dictionary(
+ train_features, test_features, train_labels, test_labels, train_weights, test_weights
+ ):
+ return {
+ "train_features": train_features,
+ "test_features": test_features,
+ "train_labels": train_labels,
+ "test_labels": test_labels,
+ "train_weights": train_weights,
+ "test_weights": test_weights,
+ }
+
+ def setUp(self):
+ super().setUp()
+ self.index = pd.Index(range(self.ROWS))
+ # `_known_at_lookahead` reads the availability columns from the UNFILTERED frame, so
+ # that is where they have to live. With no availability column present it returns
+ # None, the guard logs once and skips its second comparison entirely — so the horizon
+ # boundary and the known_at boundary are exercised from separate fixtures below.
+ self.unfiltered = frame(self.index)
+ self.features = frame(self.index, f=np.arange(self.ROWS, dtype=float))
+ self.labels = frame(self.index, y=np.arange(self.ROWS, dtype=float))
+ self.weights = SampleWeightInputs(
+ base=np.ones(self.ROWS),
+ label=None,
+ label_weighting_config=WEIGHTING_CONFIG,
+ )
+
+ def _split(self, unfiltered: pd.DataFrame, horizon: int = 0, **params):
+ model = regressor(
+ data_split_parameters={"test_size": self.TEST_ROWS, "shuffle": False, **params},
+ _label_horizon_candles=lambda pair=None: horizon,
+ )
+ return model._make_train_test_split_datasets(
+ self.features, self.labels, self.weights, self._Kitchen(), unfiltered
+ )
+
+ def test_the_horizon_boundary_purges_the_training_row_exactly_at_the_cutoff(self):
+ # train_test_split without shuffle puts the LAST 10 rows in the test set, so
+ # first_test_position is 30. With a horizon of 3 the strict comparison keeps a
+ # training row only while position < 27, so row 26 is the last survivor and row 27 is
+ # purged. `<=` would keep row 27 as well, which is the off-by-one the guard exists
+ # to prevent: that row's label horizon reaches into the test window.
+ result = self._split(self.unfiltered, horizon=3)
+ kept = list(result["train_features"].index)
+ self.assertEqual(max(kept), 26)
+ self.assertNotIn(27, kept)
+ self.assertEqual(list(result["test_features"].index), list(range(30, 40)))
+
+ def test_a_zero_horizon_purges_nothing_by_position(self):
+ # The floor of the same comparison: with no horizon the last training row sits
+ # immediately before the test window and is kept, so row 29 must survive.
+ result = self._split(self.unfiltered, horizon=0)
+ self.assertEqual(list(result["train_features"].index), list(range(30)))
+
+ def test_a_row_known_exactly_at_the_first_test_row_is_purged(self):
+ # The second, independent boundary. A label that becomes available at position
+ # position + lookahead must be strictly before the first test row, so a training row
+ # at position 25 with a lookahead of 5 is available exactly at 30 and is purged.
+ # Without the availability column the whole branch is skipped and this never bites.
+ def with_lookahead(row: int, value: int):
+ values = [0] * self.ROWS
+ values[row] = value
+ return frame(self.index, **{LOOKAHEAD: values, WEIGHT_LOOKAHEAD: [0] * self.ROWS})
+
+ unfiltered = with_lookahead(25, 5)
+ result = self._split(unfiltered, horizon=0)
+ kept = list(result["train_features"].index)
+ self.assertNotIn(25, kept, "available exactly at the first test row, so it must be purged")
+ self.assertIn(24, kept)
+ # A row one candle earlier is available at 29, strictly inside the training window.
+ kept = list(self._split(with_lookahead(24, 4), horizon=0)["train_features"].index)
+ self.assertIn(24, kept)
+
+ def test_the_horizon_and_the_availability_masks_are_combined_not_replaced(self):
+ # Pinning the two boundaries one at a time proves nothing about how they combine,
+ # which is exactly how the first version of this class went wrong: the horizon case
+ # carried no availability column and the availability case used a zero horizon, so
+ # `keep_mask &=` never ran with both constraints live. Rewriting the conjunction as an
+ # assignment — `keep_mask = train_known_at_position < ...` — discards the horizon mask
+ # and left the suite green.
+ #
+ # Both directions are needed, because either one alone is satisfied by a mask that
+ # simply overwrites the other.
+ zero_lookahead = frame(
+ self.index, **{LOOKAHEAD: [0] * self.ROWS, WEIGHT_LOOKAHEAD: [0] * self.ROWS}
+ )
+
+ # Direction 1 — the horizon mask removes rows the availability mask would keep. With
+ # a horizon of 3 the horizon keeps rows below 27 while availability, at a zero
+ # lookahead, keeps everything below the first test row at 30. The conjunction must
+ # give 26; the assignment gives 29.
+ kept = list(self._split(zero_lookahead, horizon=3)["train_features"].index)
+ self.assertEqual(max(kept), 26)
+ for row in (27, 28, 29):
+ with self.subTest(discarded_by="horizon", row=row):
+ self.assertNotIn(row, kept)
+
+ # Direction 2 — the availability mask removes a row the horizon kept. A lookahead of 6
+ # on row 20 makes it available at 26, inside the window, so it survives; a lookahead
+ # of 10 on row 20 makes it available exactly at 30 and it must go even though the
+ # horizon would have kept it.
+ for value, expected in ((6, True), (10, False)):
+ lookahead = [0] * self.ROWS
+ lookahead[20] = value
+ unfiltered = frame(
+ self.index, **{LOOKAHEAD: lookahead, WEIGHT_LOOKAHEAD: [0] * self.ROWS}
+ )
+ with self.subTest(discarded_by="availability", lookahead=value):
+ self.assertEqual(
+ 20 in self._split(unfiltered, horizon=0)["train_features"].index, expected
+ )
+
+ def test_the_causal_guard_is_skipped_outside_causal_mode(self):
+ # The counterpart, so the boundary tests above are not passing for the wrong reason:
+ # with causal_mode off the same row survives.
+ lookahead = [0] * self.ROWS
+ lookahead[25] = 5
+ unfiltered = frame(self.index, **{LOOKAHEAD: lookahead, WEIGHT_LOOKAHEAD: [0] * self.ROWS})
+ model = regressor(
+ data_split_parameters={"test_size": self.TEST_ROWS, "shuffle": False},
+ _causal_mode=False,
+ _label_horizon_candles=lambda pair=None: 0,
+ )
+ result = model._make_train_test_split_datasets(
+ self.features, self.labels, self.weights, self._Kitchen(), unfiltered
+ )
+ self.assertEqual(list(result["train_features"].index), list(range(30)))
+
+
+class TimeSeriesSplitCausalPurgeTest(QaTestCase):
+ """The chronological split, whose entire body had no test at all.
+
+ `_make_timeseries_split_datasets` is reachable through the documented
+ `data_split_parameters.method='timeseries_split'`, and coverage reported 0 of its 62
+ statements executed. Its own availability comparison at
+ QuickAdapterRegressorV3.py:3108 could go from strict to inclusive, or be deleted, with
+ the suite green.
+
+ Deliberately NOT a subclass of the class above. That one's expectations are written around
+ a ten-row test tail starting at position 30, where TimeSeriesSplit's last fold here is five
+ rows starting at 35; inheriting it made five tests assert the wrong geometry and fail. The
+ two blocks are siblings with their own arithmetic, so each carries its own fixtures.
+ """
+
+ ROWS = 40
+ N_SPLITS = 3
+ TEST_SIZE = 5
+ GAP = 2
+
+ class _Kitchen:
+ pair = PAIR
+
+ @staticmethod
+ def build_data_dictionary(
+ train_features, test_features, train_labels, test_labels, train_weights, test_weights
+ ):
+ return {
+ "train_features": train_features,
+ "test_features": test_features,
+ "train_labels": train_labels,
+ "test_labels": test_labels,
+ "train_weights": train_weights,
+ "test_weights": test_weights,
+ }
+
+ def setUp(self):
+ super().setUp()
+ self.index = pd.Index(range(self.ROWS))
+ self.unfiltered = frame(self.index)
+ self.features = frame(self.index, f=np.arange(self.ROWS, dtype=float))
+ self.labels = frame(self.index, y=np.arange(self.ROWS, dtype=float))
+ self.weights = SampleWeightInputs(
+ base=np.ones(self.ROWS),
+ label=None,
+ label_weighting_config=WEIGHTING_CONFIG,
+ )
+
+ def _split(self, unfiltered: pd.DataFrame, horizon: int = 0, **params):
+ model = regressor(
+ data_split_parameters={
+ "method": "timeseries_split",
+ "n_splits": self.N_SPLITS,
+ "test_size": self.TEST_SIZE,
+ # The code refuses a gap smaller than the horizon under causal_mode, so the
+ # horizon cases widen the gap to match rather than tripping that guard.
+ "gap": max(self.GAP, horizon),
+ **params,
+ },
+ _label_horizon_candles=lambda pair=None: horizon,
+ )
+ return model._make_timeseries_split_datasets(
+ self.features, self.labels, self.weights, self._Kitchen(), unfiltered
+ )
+
+ def test_the_last_fold_is_the_chronological_tail(self):
+ result = self._split(self.unfiltered, horizon=0)
+ test_index = list(result["test_features"].index)
+ self.assertEqual(test_index, list(range(self.ROWS - self.TEST_SIZE, self.ROWS)))
+ self.assertTrue(max(result["train_features"].index) < min(test_index))
+
+ def test_a_row_known_at_or_after_the_first_test_row_is_purged(self):
+ # The boundary this block exists for, and the one coverage could not see at all.
+ #
+ # `gap=2` removes the two rows before the test fold from the training set BEFORE the
+ # availability comparison runs, so the last row that comparison can act on is
+ # `first_test - gap - 1` = 32. Measured: train ends at 32, test is 35..39. The lookahead
+ # is placed there for that reason — on row 33 or 34 it is indistinguishable from no
+ # lookahead at all, because those rows are already gone before the comparison.
+ last_train = self.ROWS - self.TEST_SIZE - self.GAP - 1
+ first_test = self.ROWS - self.TEST_SIZE
+
+ def with_lookahead(row: int, value: int):
+ values = [0] * self.ROWS
+ values[row] = value
+ return frame(self.index, **{LOOKAHEAD: values, WEIGHT_LOOKAHEAD: [0] * self.ROWS})
+
+ # Available at last_train, +1, +2 — all strictly before the cutoff — and exactly at it.
+ for value, expected in ((0, True), (1, True), (2, True), (3, False)):
+ with self.subTest(lookahead=value):
+ kept = list(
+ self._split(with_lookahead(last_train, value), horizon=0)[
+ "train_features"
+ ].index
+ )
+ self.assertEqual(
+ last_train in kept,
+ expected,
+ f"lookahead {value} lands at {last_train + value}, cutoff {first_test}",
+ )
+
+ def test_the_availability_comparison_is_strict_at_the_cutoff(self):
+ # The isolating case: a single row whose label becomes knowable exactly on the first
+ # test row. Relaxing `<` to `<=` keeps it, and nothing else in the suite says so.
+ last_train = self.ROWS - self.TEST_SIZE - self.GAP - 1
+ values = [0] * self.ROWS
+ values[last_train] = self.GAP + 1
+ unfiltered = frame(self.index, **{LOOKAHEAD: values, WEIGHT_LOOKAHEAD: [0] * self.ROWS})
+
+ kept = list(self._split(unfiltered, horizon=0)["train_features"].index)
+
+ self.assertNotIn(last_train, kept)
+
+
if __name__ == "__main__":
unittest.main()
Covers the config-to-object boundary: which misconfigurations are fatal and which
are downgraded with a warning, plus the hermetic Optuna sampler mapping and the
-label candle pool that feeds label selection. Requires the Freqtrade QA image.
-"""
+label candle pool that feeds label selection; requires the Freqtrade QA image."""
import json
import random
-"""Provability ranking and per-candle deduplication of historic FreqAI predictions."""
+"""Provability ranking and per-candle deduplication of historic FreqAI predictions.; requires the Freqtrade QA image."""
import unittest
from types import SimpleNamespace
series, method, "rank_extrema"
)
self.assertTrue(-1.0 <= float(minimum) <= float(maximum) <= 1.0)
- np.testing.assert_allclose(
- QuickAdapterRegressorV3.skimage_min_max(series, "otsu", "rank_extrema"),
- (-0.94853515625, 0.8),
- rtol=DISTANCE_RTOL,
- atol=0.0,
- )
+
+ def test_each_requested_threshold_method_yields_its_own_numbers(self):
+ # A range check cannot tell delegation from substitution: every method lands inside
+ # [-1, 1] whatever it computes. Forcing `threshold_func` to Otsu inside the consumer
+ # left the whole suite green while `mean` silently became Otsu's -0.9485 instead of
+ # -0.55, so an entry gate used a method the config never asked for.
+ #
+ # The full tuple is pinned per method, not just the differing one. On this fixture
+ # `yen` happens to coincide with `otsu`, and `li` with `mean`; that is why the whole
+ # table is asserted — rerouting any single method to any other breaks it.
+ series = pd.Series([0.1, -0.5, 0.9, -0.2, 0.3, 0.05, -0.95, 0.7])
+ expected = {
+ "mean": (-0.55, 0.8),
+ "isodata": (-0.64970703125, 0.8),
+ "li": (-0.55, 0.8),
+ "minimum": (-0.94267578125, 0.8),
+ "otsu": (-0.94853515625, 0.8),
+ "triangle": (-0.94560546875, 0.8),
+ "yen": (-0.94853515625, 0.8),
+ }
+ self.assertEqual(sorted(expected), sorted(SKIMAGE_THRESHOLD_METHODS))
+ for method, bounds in expected.items():
+ with self.subTest(method=method):
+ np.testing.assert_allclose(
+ QuickAdapterRegressorV3.skimage_min_max(series, method, "rank_extrema"),
+ bounds,
+ rtol=DISTANCE_RTOL,
+ atol=0.0,
+ )
def test_an_unknown_skimage_threshold_is_named_in_the_canonical_enum_error(self):
with self.assertRaises(ValueError) as caught:
# ------------------------------------------------------------------ scalar validators
- def test_a_non_finite_scalar_is_refused_in_warn_and_raise_modes_and_silently_dropped_in_none(
- self,
- ):
+ def test_a_non_finite_scalar_is_refused_in_raise_mode(self):
for value in (np.inf, -np.inf, np.nan):
- with self.subTest(value=value), self.assertRaisesRegex(ValueError, "must be finite"):
+ with (
+ self.subTest(value=value, mode="raise"),
+ self.assertRaisesRegex(ValueError, "must be finite"),
+ ):
QuickAdapterRegressorV3._validate_scalar(value, ctx="label_p_order", mode=RAISE)
+
+ def test_a_non_finite_scalar_is_warned_and_dropped_in_warn_mode(self):
+ for value in (np.inf, -np.inf, np.nan):
+ with self.subTest(value=value, mode="warn"):
with self.assertLogs(LOGGER_NAME, "WARNING") as captured:
warned = QuickAdapterRegressorV3._validate_scalar(
value, ctx="label_p_order", mode=WARN
)
self.assertIsNone(warned)
self.assertIn("must be finite", captured.output[0])
- with self.assertNoLogs(LOGGER_NAME, "WARNING"):
- self.assertIsNone(
- QuickAdapterRegressorV3._validate_scalar(
- value, ctx="label_p_order", mode=NONE
- )
- )
+
+ def test_a_non_finite_scalar_is_silently_dropped_in_none_mode(self):
+ for value in (np.inf, -np.inf, np.nan):
+ with self.subTest(value=value, mode="none"), self.assertNoLogs(LOGGER_NAME, "WARNING"):
+ self.assertIsNone(
+ QuickAdapterRegressorV3._validate_scalar(value, ctx="label_p_order", mode=NONE)
+ )
+
+ # These three used to share one `with` block behind the raising call, so the warn and
+ # none branches never executed at all: the raise on the first line unwound the block.
+ # The suite was green and the test name claimed all three modes. Removing the non-finite
+ # warning in `_validate_scalar`, or turning its `return None` into `return 0.0`, both
+ # survived — the first because the warning was never asserted, the second because the
+ # fallback case in the file used a degenerate row below the ideal count.
def test_a_predicate_violation_is_reported_with_the_supplied_constraint(self):
with self.assertRaisesRegex(ValueError, "must be > 0"):
def _compose(self, base, labels):
return QuickAdapterRegressorV3._compose_eval_weights(base, labels, context=CONTEXT)
- def test_eval_weights_have_unit_mean(self):
+ def test_eval_weights_are_proportional_to_the_label_weights(self):
+ # Asserting only the unit mean is satisfied by any positive vector, including one
+ # that has thrown the label weights away and binarised the positives to all-ones —
+ # which has mean 1.0 too and shipped green. The uneven fixture is the whole point:
+ # measured, the labels [3, 1, 2, 1] give [1.714, 0.571, 1.143, 0.571] while the
+ # binarised version gives [1, 1, 1, 1].
result = self._compose(np.ones(4), np.array([3.0, 1.0, 2.0, 1.0]))
+
self.assertAlmostEqual(float(np.mean(result)), 1.0)
+ np.testing.assert_allclose(
+ result / result[0], np.array([1.0, 1.0 / 3.0, 2.0 / 3.0, 1.0 / 3.0]), rtol=1e-12
+ )
+ self.assertFalse(np.allclose(result, np.ones(4)), "the labels were discarded")
def test_eval_weights_ignore_the_support_policy_entirely(self):
# Every training threshold is unreachable here, and nothing is raised.
def test_the_fallback_policy_turns_a_support_failure_into_base_weights(self):
with self.assertLogs(
"quickadapter.user_data.freqaimodels.QuickAdapterRegressorV3", "WARNING"
- ):
+ ) as captured:
result = self._enforce(
np.ones(4),
np.ones(4),
min_effective_sample_size=4.5,
)
self.assertAlmostEqual(float(np.mean(result)), 1.0)
+ # assertLogs alone only proves THAT a warning fired. Under `fallback` the operator
+ # has no exception to read, so the warning is the whole diagnostic: it must name the
+ # threshold that actually failed, not merely announce that something did.
+ self.assertEqual(len(captured.records), 1)
+ self.assertIn("effective_sample_size=4 < min_effective_sample_size=4.5", captured.output[0])
+ self.assertNotIn("pivot_equivalent_count=0", captured.output[0])
class ComposeTrainWeightsTest(QaTestCase):
self.assertTrue(
np.isnan(model._calculate_candle_threshold(flat, PAIR, side, 0.0, 1.0))
)
+ # Checked HERE, while this frame's signature is still the live one. Every
+ # probe below rotates the timestamp on purpose, so a refusal that memoised
+ # would be wiped by the next invalidation and never observable at the end.
+ # Inserting a memo into this branch left the suite green for that reason.
+ self.assertEqual(model._candle_threshold_cache, {})
# Every probe below needs a signature the cache has not seen, because the memoised
# entry of an earlier frame shadows the branch under test: same-signature frames
start=FIRST_CANDLE + datetime.timedelta(minutes=STEP_MINUTES * offset),
)
+ # Each refusal is checked IMMEDIATELY. Asserting only at the end cannot work here:
+ # every probe rotates the frame timestamp on purpose, and
+ # `_invalidate_pair_caches` wipes the pair's entries on a signature change, so by the
+ # final assertion only the last probe's signature survives and an earlier refusal
+ # that had memoised is invisible. Making the two NaN refusal branches memoise left
+ # the suite green for exactly that reason.
for offset, natr in ((1, np.nan), (2, -1.0)):
with self.subTest(natr=natr):
deviation = model._calculate_candle_deviation(
unreadable_candle(offset, natr), PAIR, 0.0, 1.0
)
self.assertTrue(np.isnan(deviation))
+ self.assertEqual(model._candle_deviation_cache, {})
+ self.assertEqual(model._candle_threshold_cache, {})
for offset, column in ((3, "close"), (4, "open")):
holed = unreadable_candle(offset, 1.0)
holed.loc[0, column] = np.nan
self.assertTrue(
np.isnan(model._calculate_candle_threshold(holed, PAIR, side, 0.0, 1.0))
)
- # A refusal is not a value: only the zero-volatility deviation is memoised, while
- # every unreadable deviation and threshold leaves no entry behind.
+ # Only the THRESHOLD cache must stay empty here: computing a threshold
+ # legitimately memoises the deviation it derives, and that deviation is
+ # readable because the hole is in the price, not in the NATR.
+ self.assertEqual(model._candle_threshold_cache, {})
+ # And the legitimate memoisation from the flat candle is still there, so the two
+ # assertions above are not passing because the cache is simply never written to.
self.assertEqual(len(model._candle_deviation_cache), 1)
- self.assertEqual(model._candle_threshold_cache, {})
def test_the_weighted_close_lies_between_the_low_and_the_high(self):
candle = candles([110.0], opens=[100.0], highs=[120.0], lows=[95.0]).iloc[0]
lo=1.0,
hi=1.0,
)
+ # The ORDERING is a consequence of the argument validation above (lo <= hi is refused
+ # earlier), so `max >= min` holds for any input and proves nothing about the decay.
+ # What is actually observable is the historical MAXIMUM decaying alongside the
+ # minimum. Measured with these arguments, max is [0.5, 0.25] over the two follow-up
+ # calls; dropping the decay on the max gives [1.0, 1.0] and squaring the factor does
+ # too, and both survived the suite green.
for call in state["calls"]:
self.assertGreaterEqual(call["max"], call["min"])
+ self.assertAlmostEqual(state["calls"][0]["max"], 1.0)
+ self.assertAlmostEqual(state["calls"][1]["max"], 0.5)
+ self.assertAlmostEqual(state["calls"][2]["max"], 0.25)
if __name__ == "__main__":
model.get_take_profit_distance(frame(), trade(), fraction)
def test_both_distances_accept_the_closed_unit_interval(self):
+ # BOTH helpers. This case only called the stoploss one, so the take-profit path's
+ # endpoints were untested: relaxing its valid condition to `0.0 < fraction <= 1.0`,
+ # or returning None for a zero fraction before the range check, each left the suite
+ # green. The upper endpoint is covered by sibling tests in test_strategy_features;
+ # the lower one is only observable here.
+ model = strategy()
for fraction, is_zero in ((0.0, True), (1.0, False)):
with self.subTest(fraction=fraction):
- model = strategy()
- distance = model.get_stoploss_distance(frame(), trade(), 100.0, fraction)
- self.assertEqual(distance == 0.0, is_zero)
- self.assertGreaterEqual(distance, 0.0)
+ stoploss = model.get_stoploss_distance(frame(), trade(), 100.0, fraction)
+ self.assertEqual(stoploss == 0.0, is_zero)
+ self.assertGreaterEqual(stoploss, 0.0)
+ take_profit = model.get_take_profit_distance(frame(), trade(), fraction)
+ self.assertIsNotNone(take_profit, "a valid fraction must return a distance")
+ self.assertGreaterEqual(take_profit, 0.0)
+ # A zero fraction is the endpoint that can silently become an error or None.
+ self.assertEqual(model.get_take_profit_distance(frame(), trade(), 0.0), 0.0)
def test_both_distances_refuse_an_invalid_duration(self):
model = strategy(duration=0)
labelled = result[EXTREMA_COLUMN].to_numpy() != 0.0
weights = result[LABEL_WEIGHT_COLUMN]
smoothed = result["extrema_weight_smoothed"]
+ # Without this, a frame that produced no labels at all would satisfy
+ # `(weights[labelled] > 0).all()` on an empty selection. The sibling case pins
+ # the exact counts; this one only needs the selection to be non-empty.
+ self.assertTrue(labelled.any(), "no candle was labelled, so nothing was asserted")
self.assertTrue((weights[labelled] > 0.0).all())
self.assertTrue((weights >= 0.0).all())
self.assertTrue(np.isfinite(smoothed.to_numpy()).all())
- self.assertTrue((smoothed >= 0.0).all())
+
+ def test_the_labels_and_weights_do_not_depend_on_the_frame_index(self):
+ # The strongest form of the positional-contract guard, because it runs the real
+ # strategy path. Every other fixture in this suite builds `date` as a COLUMN and
+ # leaves a zero-based RangeIndex in place, where a pivot position and a pivot index
+ # label are the same number — so returning `df.index` instead of positions passed
+ # everywhere. FreqAI hands `set_freqai_targets` a DatetimeIndex frame, where the
+ # labels are Timestamps: the weighting path raised TypeError on the first candle of
+ # every pair, and a shifted integer index dropped every pivot and returned no weight
+ # at all, with only a warning.
+ baseline = None
+ for name, index in (
+ ("datetime", "date"),
+ ("shifted_int", pd.RangeIndex(500, 500 + 200)),
+ ):
+ with self.subTest(index=name), temporary_directory() as temp:
+ model = runtime(
+ temp, freqai={"label_weighting": {"default": {"strategy": "uniform"}}}
+ )
+ model.bot_start()
+ frame = zigzag().set_index(index, drop=False)
+ result = model.set_freqai_targets(frame, {"pair": PAIR})
+ labels = result[EXTREMA_COLUMN].fillna(0.0).to_numpy()
+ weights = result[LABEL_WEIGHT_COLUMN].to_numpy()
+ self.assertEqual(int((labels != 0.0).sum()), 184)
+ # 23 of the positive weights sit on pivots and the rest are the non-causal
+ # baseline fill, so this is not simply the pivot count. Pinning it means a
+ # column of zeros, or an all-positive column, cannot satisfy the assertion.
+ self.assertEqual(int((weights > 0.0).sum()), 195)
+ if baseline is None:
+ baseline = (labels, weights)
+ else:
+ np.testing.assert_array_equal(labels, baseline[0])
+ np.testing.assert_array_equal(weights, baseline[1])
def test_no_label_weight_column_appears_under_an_inactive_weighting_strategy(self):
with temporary_directory() as temp:
-"""Guards the import identity and coverage denominator of the measured source."""
+"""Guards the import identity and coverage denominator of the measured source.; requires the Freqtrade QA image."""
import importlib
import sys
QUICKADAPTER = REPO_ROOT / "quickadapter"
MEASURED_SOURCE = QUICKADAPTER / "user_data"
-_STRATEGY_SOURCES = tuple(sorted((MEASURED_SOURCE / "strategies").glob("*.py")))
-_MODEL_SOURCES = tuple(sorted((MEASURED_SOURCE / "freqaimodels").glob("*.py")))
-# Strategies modules are imported by bare name, which is what production does. The regressor is
-# not on PYTHONPATH and is reachable only as a dotted path. Deriving the names from the tree is
-# what keeps this contract true when a source file is added.
+_SOURCES = tuple(sorted(MEASURED_SOURCE.rglob("*.py")))
+# The measured tree is `quickadapter/user_data` WHOLESALE, so the contract walks it recursively:
+# a flat module such as `user_data/helpers.py` enters the denominator and must not be invisible here.
+# Strategies modules are imported by bare name, which is what production does; anything else is
+# reachable only as a dotted path under `quickadapter.user_data`. Deriving the names from the tree
+# is what keeps the contract true when a source file is added.
PRODUCTION_NAMES = tuple(
- [src.stem for src in _STRATEGY_SOURCES]
- + [f"quickadapter.user_data.freqaimodels.{src.stem}" for src in _MODEL_SOURCES]
+ src.stem
+ if src.parent.name == "strategies"
+ else src.relative_to(QUICKADAPTER.parent).with_suffix("").as_posix().replace("/", ".")
+ for src in _SOURCES
)
# A dotted strategies import is banned, so neither branch below may ever build one.
_FORBIDDEN_PREFIXES = ("quickadapter.user_data.strategies.", "quickadapter.tests.")
for module in list(sys.modules.values())
if getattr(module, "__file__", None)
}
- for src in (*_STRATEGY_SOURCES, *_MODEL_SOURCES):
+ for src in _SOURCES:
with self.subTest(module=src.name):
self.assertIn(src.resolve(), loaded)
- def test_b_every_source_module_imports(self):
- for name in PRODUCTION_NAMES:
- with self.subTest(module=name):
- self.assertIsNotNone(importlib.import_module(name))
-
def test_no_package_marker_under_quickadapter(self):
markers = sorted(
path.relative_to(QUICKADAPTER).as_posix() for path in QUICKADAPTER.rglob("__init__.py")
"""Canonical defaults contract between every `*_SPECS` map, its `DEFAULTS_*` map and the per-label overlay; requires the Freqtrade QA image."""
+import inspect
import logging
from typing import Any
+from unittest import mock
import Utils
from LabelTransformer import (
SMOOTHING_MODES,
get_label_column_config,
)
-from qa_support import QaTestCase
+from qa_support import QaTestCase, ohlcv_frame
from Utils import (
_ParamSpec,
_validate_params,
+ generate_label_data,
get_label_kind_config,
get_label_pipeline_config,
get_label_prediction_config,
)
self.assertEqual(discovered_specs(), expected)
+ def test_every_zigzag_entry_point_shares_the_canonical_natr_defaults(self):
+ # One tunable pair was written down three times with no canonical source: the
+ # params.get fallbacks in _generate_extrema_label and the two _zigzag/zigzag
+ # signatures. All three now derive from the constants, so "change the default
+ # once" is true; a per-site pin would only have been a fourth copy to drift.
+ for function in (Utils._zigzag, Utils.zigzag):
+ with self.subTest(function=function.__name__):
+ parameters = inspect.signature(function).parameters
+ self.assertEqual(parameters["natr_period"].default, Utils.DEFAULT_LABEL_NATR_PERIOD)
+ self.assertEqual(
+ parameters["natr_multiplier"].default,
+ Utils.DEFAULT_MIN_LABEL_NATR_MULTIPLIER,
+ )
+
+ def test_the_params_fallbacks_agree_with_the_canonical_constants(self):
+ # The third copy: the generator's own fallback, reached whenever a params dict
+ # omits the keys. Exercised through the public entry point.
+ with mock.patch.object(Utils, "_zigzag", wraps=Utils._zigzag) as spied:
+ generate_label_data(
+ ohlcv_frame([7.0] * 20),
+ LABEL_COLUMN,
+ params={},
+ logger=logging.getLogger("t"),
+ )
+ used = {call.kwargs["natr_period"] for call in spied.call_args_list}
+ multipliers = {call.kwargs["natr_multiplier"] for call in spied.call_args_list}
+ self.assertEqual(used, {Utils.DEFAULT_LABEL_NATR_PERIOD})
+ self.assertEqual(multipliers, {Utils.DEFAULT_MIN_LABEL_NATR_MULTIPLIER})
+
def test_a_pair_of_maps_with_disagreeing_keys_would_silently_lose_a_default(self):
# A key present in one map and absent from the other is the whole failure this
# guards: `_validate_params` reads `defaults[param]` for every spec key, so a spec
-"""Deprecated-config migration: in-place rewrite, precedence and warn-once."""
+"""Deprecated-config migration: in-place rewrite, precedence and warn-once.; requires the Freqtrade QA image."""
import logging
import unittest
# A key with nowhere to go must say why; a plain rename has nothing to explain.
self.assertEqual(new_path is None, guidance is not None)
- def test_l_section_renames_are_ordered_before_the_key_moves_that_feed_on_them(self):
- # An entry whose old path sits under an earlier entry's new path is unreachable unless
- # the earlier one has already run, which is what the ordering comment on the table
- # guarantees. Reordering the table would break every pair this finds.
- inherited = 0
- for index, (old_path, _, _, _) in enumerate(CONFIG_DEPRECATIONS):
- for earlier_index, (_, earlier_new, _, _) in enumerate(CONFIG_DEPRECATIONS[:index]):
- if earlier_new is None:
- continue
- if old_path == earlier_new or old_path.startswith(f"{earlier_new}."):
- inherited += 1
- self.assertLess(
- earlier_index,
- index,
- f"{old_path} can only be reached after {earlier_index} is applied",
- )
- self.assertGreater(inherited, 0, "the table no longer encodes a rename chain")
+ def test_a_renamed_section_reaches_its_keys_through_the_second_hop(self):
+ # The table is order-dependent by construction: entry 0 renames
+ # `freqai.extrema_weighting` to `freqai.label_weighting`, and entries 21-27 then move
+ # keys from there to `freqai.label_pipeline`. A key written under the ORIGINAL section
+ # therefore only lands in its final home if the section rename has already run.
+ #
+ # The previous version of this test asserted `earlier_index < index` with
+ # `earlier_index` drawn from `enumerate(CONFIG_DEPRECATIONS[:index])`, which is true by
+ # construction, so reordering the table kept it green while silently stranding the key.
+ config = {
+ "freqai": {
+ "extrema_weighting": {
+ "normalization": "minmax",
+ "gamma": 2.0,
+ }
+ }
+ }
+
+ migrate_config(config, LOGGER)
+
+ self.assertEqual(
+ config["freqai"]["label_pipeline"],
+ {"normalization": "minmax", "gamma": 2.0},
+ )
+ self.assertNotIn("extrema_weighting", config["freqai"])
+
+ def test_the_table_still_encodes_a_rename_chain(self):
+ # The structural half, which is a fact about the table rather than about behaviour:
+ # at least one entry's old path must sit under an earlier entry's new path, or the
+ # behavioural test above has stopped testing a chain.
+ inherited = [
+ old_path
+ for old_path, _, _, _ in CONFIG_DEPRECATIONS
+ for _, earlier_new, _, _ in CONFIG_DEPRECATIONS
+ if earlier_new is not None
+ and (old_path == earlier_new or old_path.startswith(f"{earlier_new}."))
+ ]
+ self.assertGreater(len(inherited), 0, "the table no longer encodes a rename chain")
+
+ def test_a_two_hop_rename_reaches_its_final_key(self):
+ # The chain test above covers ONE section rename plus a key move. The table also
+ # contains two-hop chains inside a single section, and asserting that "a chain exists"
+ # plus testing one of them says nothing about the others.
+ # `threshold_outlier` -> `outlier_threshold_quantile` -> `outlier_quantile`
+ for old_key, expected in (("threshold_outlier", "outlier_quantile"),):
+ with self.subTest(key=old_key):
+ config = {"freqai": {"label_prediction": {old_key: 0.03}}}
+ migrate_config(config, LOGGER)
+ self.assertEqual(config["freqai"]["label_prediction"], {expected: 0.03})
+
+ # `extrema_fraction` -> `keep_extrema_fraction` -> `keep_fraction`
+ config = {"freqai": {"label_prediction": {"extrema_fraction": 0.5}}}
+ migrate_config(config, LOGGER)
+ self.assertEqual(config["freqai"]["label_prediction"], {"keep_fraction": 0.5})
+
+ def test_a_key_under_a_renamed_prediction_section_reaches_its_final_home(self):
+ # Entry 2 renames `freqai.predictions_extrema` to `freqai.label_prediction`, and
+ # entries 5-10 then move its keys. Sorted by old path, the key move would run BEFORE
+ # the section rename and strand the key: the section's deprecated key survives
+ # untouched and the threshold silently reverts to its default.
+ config = {"freqai": {"predictions_extrema": {"thresholds_smoothing": "mean"}}}
+
+ migrate_config(config, LOGGER)
+
+ self.assertEqual(config["freqai"]["label_prediction"], {"threshold_method": "mean"})
+ self.assertNotIn("predictions_extrema", config["freqai"])
def test_m_a_deprecation_path_warns_once_per_process_across_config_objects(self):
first = {"freqai": {"extrema_smoothing": {"factor": 0.5}}}
-"""Strategy-facing config resolvers: defaults precedence, warning text, and per-call isolation."""
+"""Strategy-facing config resolvers: defaults precedence, warning text, and per-call isolation.; requires the Freqtrade QA image."""
import copy
import logging
logger = logging.getLogger("quickadapter.tests.config_resolution")
handler = _WarningSink(sink)
previous_propagate = logger.propagate
+ previous_level = logger.level
logger.propagate = False
- logger.setLevel(logging.NOTSET)
+ # The level is set explicitly AND restored. NOTSET makes the logger inherit the root
+ # level, and a record the logger has already filtered out never reaches a private
+ # handler — so under a root at ERROR every case asserting on warning text fails on an
+ # empty sink. Restoring it on exit matters too: leaving it changed makes later tests
+ # depend on ordering.
+ logger.setLevel(logging.WARNING)
logger.addHandler(handler)
try:
yield logger, sink
finally:
logger.removeHandler(handler)
logger.propagate = previous_propagate
+ logger.setLevel(previous_level)
class UtilsConfigResolutionTest(QaTestCase):
-"""Indicator kernel contracts: extrema counting, log returns, smoothing and the callback hash."""
+"""Indicator kernel contracts: extrema counting, log returns, smoothing and the callback hash.; requires the Freqtrade QA image."""
import functools
import hashlib
from Utils import (
MA_MODES,
PRICE_MODES,
+ _fractal_dimension,
alligator,
bottom_log_return,
calculate_min_extrema,
self.assertTrue(np.isnan(result[:5]).all())
assert_allclose(result[5:], np.full(15, 7.0), rtol=0.0, atol=0.0)
- def test_frama_recurses_toward_close_on_a_series_with_a_real_range(self):
- # The other frama fixtures are all CONSTANT frames, and _fractal_dimension returns
- # 1.0 whenever either half-range is zero. So alpha collapses to exp(-4.6 * 0) == 1.0
- # and neither the -4.6 scale, nor the `fd - 1` offset, nor the clip bounds are
- # observable: a mutant that doubles alpha survives on all of them. Only a frame
- # with a genuine high/low range in every window can see any of the three.
- result = frama(ohlcv_frame([100.0 + i for i in range(30)]), 6).to_numpy()
+ def test_frama_reproduces_close_because_the_dimension_is_one_up_to_rounding(self):
+ # The real contract, asserted rather than implied. HL1 and HL2 each span a HALF of the
+ # window while HL3 spans all of it, so HL1 + HL2 <= 2 * HL3 and D <= 1, which makes
+ # alpha exp(-4.6 * (D - 1)) indistinguishable from 1 at this tolerance. This pins the
+ # DEGENERACY as a fact: a change that made the dimension vary, or the -4.6 scale
+ # observable, would have to update this test rather than pass unnoticed. It does NOT
+ # establish that alpha is a literal 1.0 — this fixture is monotone and happens to land
+ # on exactly 1.0, where
+ # test_the_fractal_dimension_can_exceed_one_by_rounding does not.
+ closes = np.array([100.0 + i for i in range(30)])
+ result = frama(ohlcv_frame(closes.tolist()), 6).to_numpy()
self.assertTrue(np.isnan(result[:5]).all())
- # Derivation of the tail: the window seeds from the mean of the first `period`
- # closes (102.5) and each subsequent bar smooths toward the previous one, which on
- # this unit ramp leaves close minus 0.5 after the first jump.
- assert_allclose(result[5], 102.5, rtol=0.0, atol=1e-9)
- assert_allclose(result[6], 106.0, rtol=0.0, atol=1e-9)
- assert_allclose(result[7:], np.arange(107.0, 130.0), rtol=0.0, atol=1e-9)
+ assert_allclose(result[5], closes[:6].mean(), rtol=0.0, atol=1e-9)
+ assert_allclose(result[6:], closes[6:], rtol=0.0, atol=1e-9)
+
+ def test_the_fractal_dimension_is_one_within_rounding_on_every_window(self):
+ # Asserted directly on the helper. The name says "within rounding" on purpose: the
+ # clip's lower bound is not a floor in practice, because `np.clip` is a no-op for
+ # values already at or above it, and the ratio lands on 1.0 + O(1e-16) for some
+ # windows. Asserting `== 1.0` here would be pinning a rounding accident of these
+ # three fixtures rather than the contract.
+ trending_high = 100.0 + np.arange(16) * 0.5
+ trending_low = trending_high - 0.8
+ noisy = np.array([100.0 + i for i in range(16)])
+ plateaus = np.array(
+ [1.0, 1.0, 1.0, 5.0, 5.0, 5.0, 2.0, 2.0, 2.0, 9.0, 9.0, 9.0, 3.0, 3.0, 3.0, 4.0]
+ )
+ for highs, lows in (
+ (trending_high, trending_low),
+ (noisy, noisy - 0.3),
+ (plateaus, plateaus - 1.0),
+ ):
+ with self.subTest(highs=highs[:3].tolist()):
+ self.assertAlmostEqual(_fractal_dimension(highs, lows, 16), 1.0, places=12)
+
+ def test_the_fractal_dimension_can_exceed_one_by_rounding(self):
+ # The counterexample that makes the case above honest. This window is not synthetic:
+ # the ratio computes to 1.0000000000000002, the clip leaves it untouched because it is
+ # already above the lower bound, and alpha becomes 0.999999999999999. Anything that
+ # branches on `alpha == 1.0` or promises bit-exact `close` is wrong on this input.
+ highs = np.array([100.2, 101.2, 102.7, 101.6, 102.8, 102.5, 102.5, 102.8, 100.9, 100.3])
+ lows = np.array([99.9, 100.9, 101.6, 101.1, 102.4, 102.3, 101.6, 101.7, 100.6, 99.9])
+ dimension = _fractal_dimension(highs, lows, 10)
+ self.assertGreater(dimension, 1.0)
+ self.assertLessEqual(dimension, 1.0 + 1e-15)
+ self.assertLess(np.exp(-4.6 * (dimension - 1.0)), 1.0)
def test_the_zero_lag_frama_de_lags_before_the_fractal_dimension(self):
# zero_lag replaces high, low and close with calculate_zero_lag first, which makes the
)
self.assertAlmostEqual(float(raw[3]), float(closes.iloc[:4].mean()), places=9)
- def test_a_framed_gap_stops_the_fractal_dimension_at_the_window_edge(self):
- # _fractal_dimension returns 1.0 when either half-range is zero, which pins alpha at
- # exp(0) = 1 and makes the recursion follow close exactly. A constant frame is the
- # only shape that reaches that branch.
+ def test_a_constant_window_gives_a_finite_fractal_dimension_instead_of_a_log_of_zero(self):
+ # _fractal_dimension short-circuits to 1.0 when either half-range is zero, and a
+ # constant frame is the only shape that reaches that branch. The case was previously
+ # named for a gap and a window edge that its fixture cannot contain, which read as if
+ # the early return were covered from a second direction. What is actually distinct
+ # here is the guard itself: without it the ratio is log(0) and alpha is -inf.
result = frama(ohlcv_frame([7.0] * 12), 4).to_numpy()
self.assertFalse(np.isinf(result).any())
- assert_allclose(result[3:], np.full(9, 7.0), rtol=0.0, atol=0.0)
+ self.assertFalse(np.isnan(result[3:]).any())
def test_the_smma_follows_its_own_recursion(self):
# Seeded with the mean of the first `period` bars, then alpha = 1/period per bar.
with self.subTest(period=period), self.assertRaisesRegex(ValueError, "must be even"):
frama(frame, period)
+ def test_a_fractal_dimension_period_below_two_is_refused(self):
+ # The `period < 2` clause is a separate contract from the evenness clause, and the
+ # existing case only drives odd periods, which the evenness clause already catches.
+ # Zero is even. Without its own case the clause can be dropped and the caller gets
+ # numpy's internal "zero-size array to reduction operation maximum which has no
+ # identity" instead of the documented domain error.
+ frame = ohlcv_frame([1.0, 2, 3, 4, 5, 6])
+ for period in (0, -4):
+ with (
+ self.subTest(period=period),
+ self.assertRaisesRegex(ValueError, "must be an even integer >= 2"),
+ ):
+ frama(frame, period)
+
def test_the_alligator_lines_are_the_shifted_smoothed_averages_of_the_median_price(self):
# The multipliers are asymmetric so that median, close, average, typical and
# weighted-close are five different series: a symmetric frame collapses them all
import logging
import unittest
+from typing import ClassVar
import numpy as np
import pandas as pd
from Utils import (
LabelWeightSupportError,
_effective_sample_size,
+ compose_label_lookahead,
compose_sample_weights,
+ compute_label_weights,
nan_average,
non_zero_diff,
sanitize_and_renormalize,
self.assertTrue(np.isnan(result.iloc[0]))
+class ComposeLabelLookaheadTest(QaTestCase):
+ """The right kernel boundary, which `compose_label_lookahead` has to hold shut.
+
+ A centered smoothing kernel needs future candles that do not exist yet, so the last
+ `kernel_half_width` rows must stay unavailable until the frame grows. The bound is
+ `right_edge_start = max(0, n - kernel_half_width)` and it was untested: widening it by a
+ single row kept the whole suite green while changing the reported lookahead for that row
+ from `n - row` to 1, on every combination the function is reachable with.
+ """
+
+ def test_the_last_kernel_half_width_rows_stay_unavailable(self):
+ n, half = 300, 5
+ # A known_at of 0 everywhere, so the available position equals the row itself. The
+ # centered rolling max then returns `last row of the window` for every row, which
+ # would leave the final rows available immediately: at row n-1 the window ends at
+ # n-1, giving a lookahead of 0. The clamp holds those rows to `n` instead, so the
+ # reported lookahead is exactly `n - row` and decays to 1 at the last row.
+ result = compose_label_lookahead(
+ pd.Series(np.zeros(n, dtype=np.int64)), half, method="gaussian", mode="zero"
+ )
+
+ # The clamped window, and only it, is pinned to `n`.
+ np.testing.assert_array_equal(result.to_numpy()[n - half :], np.arange(half, 0, -1))
+ self.assertEqual(int(result.iloc[-1]), 1)
+ # One row earlier the rolling max already returns 5, not the 4 the clamp would give,
+ # so the boundary is exactly where the clamp starts and not one row off.
+ self.assertEqual(int(result.iloc[n - half - 1]), half)
+
+ def test_a_zero_half_width_returns_the_input_untouched(self):
+ values = np.array([0, 3, 1, 7], dtype=np.int64)
+ result = compose_label_lookahead(pd.Series(values), 0, method="gaussian", mode="zero")
+ np.testing.assert_array_equal(result.to_numpy(), values)
+
+
+class MetricWeightingStrategyTest(QaTestCase):
+ """The per-metric weighting strategies, which no test exercised.
+
+ Only the `uniform` strategy was ever selected, so `weights = np.asarray(metrics[strategy])`
+ never ran: replacing it with `np.ones` left the whole suite green while every configured
+ swing importance silently became uniform. The amplitude strategy is reachable from a
+ real config and is the alternative the shipped template documents.
+ """
+
+ METRICS: ClassVar[dict[str, list[float]]] = {
+ "amplitude": [0.2, 0.7],
+ "speed": [1.0, 4.0],
+ "efficiency_ratio": [0.5, 0.9],
+ }
+
+ def _weights(self, strategy: str) -> np.ndarray:
+ return compute_label_weights(
+ 10, [2, 8], self.METRICS, {"strategy": strategy}, logger=LOGGER
+ )
+
+ def test_each_metric_strategy_places_that_metric_on_the_pivots(self):
+ for strategy, values in self.METRICS.items():
+ with self.subTest(strategy=strategy):
+ weights = self._weights(strategy)
+ self.assertEqual(weights[2], values[0])
+ self.assertEqual(weights[8], values[1])
+ # Everything that is not a pivot carries no label weight at all.
+ off_pivot = np.delete(weights, [2, 8])
+ np.testing.assert_array_equal(off_pivot, np.zeros(8))
+
+ def test_a_metric_strategy_keeps_the_relative_importance_of_the_pivots(self):
+ # The part the consumer actually sees. `compose_sample_weights` is what turns these
+ # into the training sample weights, so it is the ratio there that must survive a
+ # uniformisation of the metric — mean, and the fact that weights are positive, hold
+ # either way.
+ for strategy, expected in (("amplitude", 3.5), ("speed", 4.0), ("efficiency_ratio", 1.8)):
+ with self.subTest(strategy=strategy):
+ weights = self._weights(strategy)
+ labels = np.zeros(10)
+ labels[[2, 8]] = weights[[2, 8]]
+ composed = compose_sample_weights(
+ np.ones(10), labels, logger=LOGGER, context=CONTEXT
+ )
+ self.assertAlmostEqual(float(composed[8] / composed[2]), expected, places=9)
+
+ def test_uniform_still_ignores_the_metrics(self):
+ np.testing.assert_array_equal(
+ compute_label_weights(10, [2, 8], self.METRICS, {"strategy": "uniform"}, logger=LOGGER)[
+ [2, 8]
+ ],
+ np.ones(2),
+ )
+
+
if __name__ == "__main__":
unittest.main()
def test_an_exact_multiple_is_its_own_floor_and_ceiling(self):
self.assertEqual(floor_to_step(9, 3), 9)
self.assertEqual(ceil_to_step(9, 3), 9)
- self.assertEqual(round_to_step(9, 3), 9)
+
+ def test_a_fractional_value_is_snapped_before_it_is_truncated(self):
+ # Every other case in this class passes a Python int, so the float branch of
+ # `_step_round` — which divides the value as a float, rounds the quotient, and
+ # multiplies the step back — was never executed, and coverage reported its lines
+ # missing. The exact-multiple cases cannot tell it from the identity either.
+ #
+ # 4.6 is the case that separates the two readings: rounding the quotient gives
+ # round(4.6 / 3) == 2, hence 6, while truncating the value to an integer first gives
+ # round(4 / 3) == 1, hence 3. A step below 1 is not available to widen the gap —
+ # `_validate_step_args` requires a positive integer.
+ self.assertEqual(round_to_step(4.6, 3), 6)
+ self.assertNotEqual(round_to_step(4.6, 3), round_to_step(4, 3))
+ # The same for the negative direction, where truncation moves the other way.
+ self.assertEqual(round_to_step(-4.6, 3), -6)
+ self.assertEqual(floor_to_step(4.6, 3), 3)
+ self.assertEqual(ceil_to_step(4.6, 3), 6)
+ # The step is always a multiple, whatever the input type, and the result is an int.
+ for value in (4.6, -4.6, 4.4, 0.5, 2.5, 7.0, 8.0):
+ for function in (round_to_step, floor_to_step, ceil_to_step):
+ with self.subTest(function=function.__name__, value=value):
+ self.assertEqual(function(value, 3) % 3, 0)
+ self.assertIsInstance(function(value, 3), int)
def test_a_non_positive_step_is_refused(self):
for step in (0, -1):
)
self.assertNotEqual(result.volume_rates[0], 1.0)
+ def test_the_volume_weighted_efficiency_ratio_weights_each_step_by_its_destination(self):
+ # The sixth COMBINED_METRIC, and the only one that had no numeric assertion: every
+ # existing check either mirrored `efficiency_ratio` or expected 1.0, which holds for
+ # ANY positive weighting. The W_LEGS fixture is monotone, so its ratio is 1.0 under
+ # every convention and cannot discriminate. Shifting the volume window by one candle
+ # — same length, different slice — left all 968 tests green.
+ #
+ # This fixture is a slow uptrend carrying a fast oscillation: the legs are short and
+ # genuinely non-monotone, so the ratio drops well below 1.0 and the window offset
+ # changes the answer. The oracle below restates the documented convention: every
+ # close-to-close step is weighted by the volume of the candle it lands ON, and the
+ # pivot bar at the start of the leg is excluded from the total.
+ steps = np.arange(120)
+ closes = 100.0 + 0.02 * steps + 0.25 * np.sin(steps / 2.0)
+ volumes = 10.0 + 5.0 * np.sin(steps / 4.0)
+
+ result = _zigzag(ohlcv_frame(closes, volumes), **PARAMS)
+
+ log_closes = np.log(closes)
+ checked = 0
+ for index in range(len(result.indices) - 1):
+ start, end = sorted((result.indices[index], result.indices[index + 1]))
+ end += 1
+ window = volumes[start + 1 : end]
+ weighted = np.diff(log_closes[start:end]) * (window / window.sum())
+ expected = abs(weighted.sum()) / np.abs(weighted).sum()
+ with self.subTest(leg=index):
+ self.assertAlmostEqual(
+ float(result.volume_weighted_efficiency_ratios[index]),
+ float(expected),
+ places=12,
+ )
+ checked += 1
+ self.assertGreater(checked, 5, "the fixture must produce enough legs to be meaningful")
+ # And the convention is load-bearing here, not an identity: the one-candle shift a
+ # mutation introduces, and the unweighted ratio, both differ from the pinned value.
+ start, end = sorted((result.indices[5], result.indices[6]))
+ steps_slice = np.diff(log_closes[start : end + 1])
+ shifted = np.abs(steps_slice * (volumes[start:end] / volumes[start:end].sum())).sum()
+ self.assertNotAlmostEqual(
+ float(result.volume_weighted_efficiency_ratios[5]),
+ abs(steps_slice.sum()) / shifted,
+ places=9,
+ )
+
def test_the_efficiency_ratio_measures_directness_of_the_leg(self):
closes = wiggly_closes()
result = zigzag_result(closes)
self.assertEqual(sorted(set(label.series.tolist())), [-1.0, 0.0, 1.0])
self.assertEqual(label.indices, [0, 29, 69])
for position, direction in zip(label.indices, [1.0, -1.0, 1.0], strict=True):
- self.assertEqual(label.series.loc[position], direction)
+ self.assertEqual(label.series.iloc[position], direction)
self.assertEqual(int((label.series != 0.0).sum()), len(label.indices))
+ def test_pivot_indices_are_positions_not_index_labels(self):
+ # `LabelData.indices` is documented as "positions of detected pivots in series", and
+ # `compute_label_weights` consumes it that way: it casts to int and bounds the result
+ # by the row count. Every other fixture in this suite uses a zero-based RangeIndex,
+ # where a label and a position coincide, so returning `df.index` instead of positions
+ # would pass everywhere. These three indexes are what make the two distinguishable.
+ closes = price_path(*W_LEGS)
+ expected = generate_label_data(ohlcv_frame(closes), EXTREMA_COLUMN, PARAMS, LOGGER)
+ indexes = {
+ "shifted_int": pd.RangeIndex(500, 500 + len(closes)),
+ "datetime": pd.date_range("2024-01-01", periods=len(closes), freq="5min"),
+ "string": pd.Index([f"candle-{i}" for i in range(len(closes))]),
+ }
+ for name, index in indexes.items():
+ with self.subTest(index=name):
+ frame = ohlcv_frame(closes).set_axis(index)
+ label = generate_label_data(frame, EXTREMA_COLUMN, PARAMS, LOGGER)
+ self.assertEqual(label.indices, expected.indices)
+ self.assertEqual(label.indices, sorted(set(label.indices)))
+ np.testing.assert_array_equal(label.series.to_numpy(), expected.series.to_numpy())
+ self.assertTrue(label.series.index.equals(index))
+
+ def test_label_weights_survive_an_index_that_is_not_a_position(self):
+ # The consumer side of the same contract. `compute_label_weights` casts the pivot
+ # indices to int and bounds them by n_values, so index labels raise TypeError on the
+ # DatetimeIndex that FreqAI actually passes, and are dropped as out-of-range on a
+ # shifted integer index, leaving every weight at zero with only a warning.
+ from Utils import compute_label_weights
+
+ closes = price_path(*W_LEGS)
+ expected = generate_label_data(ohlcv_frame(closes), EXTREMA_COLUMN, PARAMS, LOGGER)
+ for name, index in (
+ ("shifted_int", pd.RangeIndex(500, 500 + len(closes))),
+ ("datetime", pd.date_range("2024-01-01", periods=len(closes), freq="5min")),
+ ):
+ with self.subTest(index=name):
+ frame = ohlcv_frame(closes).set_axis(index)
+ label = generate_label_data(frame, EXTREMA_COLUMN, PARAMS, LOGGER)
+ weights = compute_label_weights(
+ len(closes),
+ label.indices,
+ {"efficiency_ratio": [1.0] * len(label.indices)},
+ {"strategy": "uniform"},
+ logger=LOGGER,
+ )
+ self.assertEqual(int((weights > 0).sum()), len(expected.indices))
+
def test_the_label_metrics_mirror_the_pivot_metrics(self):
frame = ohlcv_frame(price_path(*W_LEGS))
result = _zigzag(frame, **PARAMS)
[-1] * (int(unresolved.sum()) - 1),
)
+ def test_a_pivot_is_never_available_before_its_confirmation_candle(self):
+ # The monotonic watermark in `add_pivot` takes the max of four terms, and the one
+ # this pins is `confirmed_at_pos`. On W_LEGS the orientation candle and the ATR
+ # warm-up end coincide, so `max(confirmed, 0, warmup, -1)` equals
+ # `max(0, warmup, -1)` and dropping the term changes nothing — which is why the
+ # case above cannot tell the two apart.
+ #
+ # A long flat stretch then a trend separates them: the ATR is finite from the fifth
+ # candle, but the orientation at row 0 is not confirmed until the trend has produced
+ # five consecutive slopes. Dropping `confirmed_at_pos` then publishes the label of
+ # row 0 long before its confirmation candle. Measured over seeded frames, that term
+ # is the difference on 509 of 1500 and makes a row available before its own candle on
+ # 41 of them, by up to 96 candles.
+ quiet = 40
+ closes = np.concatenate([np.full(quiet, 100.0), 100.0 + np.arange(1, 61) * 0.8])
+ frame = ohlcv_frame(closes)
+
+ result = _zigzag(frame, **PARAMS)
+ label = generate_label_data(frame, EXTREMA_COLUMN, PARAMS, LOGGER)
+
+ self.assertEqual(result.indices, [0])
+ self.assertEqual(int(result.known_at_positions[0]), quiet)
+ # And the separation is real, not a coincidence: the ATR warm-up term alone spans a
+ # handful of candles, so a value of `quiet` can only come from the confirmation.
+ self.assertGreater(int(result.known_at_positions[0]), 4 * NATR_PERIOD)
+ self.assertEqual(int(label.known_at_lookahead.iloc[0]), quiet)
+ self.assertTrue(np.all(result.known_at_positions >= np.arange(len(closes), dtype=np.int64)))
+
def test_an_unregistered_label_column_raises_a_key_error_listing_the_registry(self):
frame = ohlcv_frame(price_path(*W_LEGS))
params: dict[str, Any],
logger: Logger | None = None,
) -> LabelData:
- natr_period = params.get("natr_period", 14)
- natr_multiplier = params.get("natr_multiplier", 9.0)
+ natr_period = params.get("natr_period", DEFAULT_LABEL_NATR_PERIOD)
+ natr_multiplier = params.get("natr_multiplier", DEFAULT_MIN_LABEL_NATR_MULTIPLIER)
result = _zigzag(
dataframe,
natr_period=natr_period,
series = pd.Series(0.0, index=dataframe.index)
if result.indices:
- series.loc[result.indices] = result.directions
+ # `indices` are POSITIONS and `series` carries the caller's index, so this assignment
+ # must be positional. `.loc` would read the positions as labels, which raises KeyError
+ # on the DatetimeIndex FreqAI passes and on any shifted integer index, and on a
+ # PERMUTED integer index writes the directions onto the wrong rows without raising.
+ series.iloc[result.indices] = result.directions
metrics: dict[str, list[float]] = {
COMBINED_METRICS[0]: result.amplitudes, # "amplitude"
DEFAULT_MIN_LABEL_PERIOD_CANDLES: Final[int] = 12
DEFAULT_MAX_LABEL_PERIOD_CANDLES: Final[int] = 24
DEFAULT_MIN_LABEL_NATR_MULTIPLIER: Final[float] = 9.0
+DEFAULT_LABEL_NATR_PERIOD: Final[int] = 14
DEFAULT_MAX_LABEL_NATR_MULTIPLIER: Final[float] = 12.0
def _fractal_dimension(
highs: NDArray[np.floating], lows: NDArray[np.floating], period: int
) -> float:
- """Original fractal dimension computation implementation per Ehlers' paper."""
- if period % 2 != 0:
- raise ValueError(f"Invalid period value {period!r}: must be even")
+ """Fractal dimension over the two half-window ranges of a ``period`` window.
+
+ ``HL3`` spans the whole window while ``HL1`` and ``HL2`` each span a half, so
+ ``HL1 <= HL3`` and ``HL2 <= HL3``, hence ``HL1 + HL2 <= 2 * HL3`` and
+ ``D <= 1`` for every reachable input. In exact arithmetic the clip would then
+ return ``1.0`` on every path. In floating point it does not: the ratio lands
+ on ``1.0 + O(1e-16)`` for some windows, and ``np.clip(D, 1.0, 2.0)`` is a
+ no-op above the lower bound, so it passes that value through. ``alpha`` in
+ :func:`frama` is therefore ``exp(-4.6 * (D - 1))``, which is ``1.0`` to within
+ a few ULP rather than exactly ``1.0``, and ``frama`` reproduces ``close``
+ after its seed up to that same rounding. Measured on one deterministic
+ 10-bar window, ``D`` is ``1.0000000000000002`` and ``alpha`` is
+ ``0.999999999999999``.
+
+ Nothing downstream may treat ``alpha`` as a literal ``1.0`` or ``frama`` as
+ bit-exact ``close``; a short-circuit on ``alpha == 1`` would not fire.
+
+ This is the current, asserted behaviour, not the formulation in Ehlers' paper:
+ that one takes the range over ``period // 2`` consecutive sub-ranges, whose sum
+ can exceed twice the full range. Adopting it is an algorithmic change and is
+ deliberately not made here.
+ """
+ if period % 2 != 0 or period < 2:
+ raise ValueError(f"Invalid period value {period!r}: must be an even integer >= 2")
half_period = period // 2
def _zigzag(
df: pd.DataFrame,
- natr_period: int = 14,
- natr_multiplier: float = 9.0,
+ natr_period: int = DEFAULT_LABEL_NATR_PERIOD,
+ natr_multiplier: float = DEFAULT_MIN_LABEL_NATR_MULTIPLIER,
normalize: bool = False,
*,
logger: Logger | None = None,
natr_warmup_end_pos = int(finite_natr_positions[0]) if finite_natr_positions.size > 0 else n
natr_values = natr.bfill().to_numpy()
- indices: list[int] = df.index.tolist()
+ # Pivots are tracked as POSITIONS, never as index labels. `compute_label_weights`
+ # consumes `ZigzagResult.indices` as row offsets (it casts to int and bounds them by
+ # `n_values`), and FreqAI hands the strategy a DatetimeIndex frame, so index labels
+ # would be Timestamps there: the weighting path raises TypeError on the first candle
+ # of every pair, and on any non-zero-based integer index it silently drops every pivot
+ # and returns all-zero weights. `LabelData.indices` documents the positional contract.
thresholds: NDArray[np.floating] = natr_values * natr_multiplier
closes = df.get("close").to_numpy(dtype=float)
highs = df.get("high").to_numpy(dtype=float)
latest_confirmation_pos = confirmed_at_pos
known_at_positions[last_resolved_pos + 1 : resolve_through_pos + 1] = confirmed_at_pos
last_resolved_pos = max(last_resolved_pos, resolve_through_pos)
- if pivots_indices and indices[pos] == pivots_indices[-1]:
+ if pivots_indices and pos == pivots_indices[-1]:
return
# These swing metrics are backfilled onto the previous pivot from the
pivots_efficiency_ratios[-1] = efficiency_ratio
pivots_volume_weighted_efficiency_ratios[-1] = volume_weighted_efficiency_ratio
- pivots_indices.append(indices[pos])
+ pivots_indices.append(pos)
pivots_values_log.append(value_log)
pivots_directions.append(direction)
def zigzag(
df: pd.DataFrame,
- natr_period: int = 14,
- natr_multiplier: float = 9.0,
+ natr_period: int = DEFAULT_LABEL_NATR_PERIOD,
+ natr_multiplier: float = DEFAULT_MIN_LABEL_NATR_MULTIPLIER,
normalize: bool = False,
*,
logger: Logger | None = None,
repositories / freqai-strategies.git / commitdiff
