"""Offline validator for the bundled v0.5 educational card and field catalogue.

Uses only the Python standard library. No network, imports into ITradingBot,
execution, or profitability evaluation. The schema reader implements only the
explicitly supported keywords below and fails on unsupported schema keywords.
"""

# (c) 2026 ITradingBot Limited. All rights reserved.
# Published so that a user's own AI assistant can read the ITradingBot backtest contract and build cards and preset links for it. Not for redistribution, and not for use in another product.
from __future__ import annotations

import argparse
import hashlib
import json
import math
import re
from collections import Counter
from datetime import datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any

HERE = Path(__file__).resolve().parent
ROLE_LIMIT_KEYS = {
    "entry": ("entry", "max_main_ind_count"),
    "filter": ("filter", "max_price_limit_ind_count"),
    "exit": ("exit", "max_close_by_ind_count"),
}
PROFILE_LIMITS = {"entry": 1, "filter": 10, "exit": 3}
DEFAULT_BINDINGS = {
    "exchange": "backtest.exchange", "asset": "backtest.symbol",
    "dca_enabled": "profile.dca_enabled", "mode": "backtest.backtest_mode",
    "start": "backtest.start_dt", "end": "backtest.end_dt",
}
MODE_DCA = {"AVERAGE_BY_LIMIT": False, "AVERAGE_BY_SIGNAL": True}
SCHEMA_KEYWORDS = {
    "$schema", "$id", "$ref", "title", "description", "definitions",
    "type", "const", "enum", "required", "properties", "additionalProperties",
    "items", "minItems", "maxItems", "uniqueItems", "minLength", "maxLength",
    "minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum",
    "allOf", "anyOf", "oneOf", "not", "if", "then", "else",
}
TYPE_ALIASES = {"str": "string", "int": "integer", "float": "number", "bool": "boolean"}
COMPLETE_STATUSES = {"complete_example", "complete_proposal"}
USER_REQUIRED_PROPOSAL_SETTINGS = {
    "backtest.symbol", "backtest.start_dt", "backtest.end_dt",
    "backtest.initial_quote_qty", "backtest.first_order",
    "backtest.leverage",
}
# Pair-list ceilings of the reviewed calculation service (source defaults): one pair without VIP,
# up to 600 with VIP. Checked when a calculation starts, never when a link is opened.
MAX_CARD_SYMBOLS = 600
SINGLE_SYMBOL_TIER_LIMIT = 1
DEPENDENCY_FINGERPRINT ='787b4b9935227e909f1dc6e9a3c68df8a73cf37ca11ad8791774c6f60603a6db'
MAIN_CONSTRAINT_FINGERPRINT = '0306b281367215cd56b607ae69bf0cda6b05c9835ac0c443c3b3c7a3eaefa08a'


def issue(path: str, code: str, message: str) -> dict[str, str]:
    return {"path": path, "code": code, "message": message}


def preset_datetime(value: str) -> datetime:
    """Exact local wall-clock minutes accepted by this browser preset profile."""
    if isinstance(value, str):
        for pattern, date_format in ((r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}", "%Y-%m-%dT%H:%M"),
                                     (r"\d{2}\.\d{2}\.\d{4} \d{2}:\d{2}", "%d.%m.%Y %H:%M")):
            if re.fullmatch(pattern, value):
                return datetime.strptime(value, date_format)
    raise ValueError("Dates require local wall-clock minutes, without seconds or timezone offsets")


def json_equal(left: Any, right: Any) -> bool:
    """JSON equality without Python's True == 1 shortcut."""
    if isinstance(left, bool) or isinstance(right, bool):
        return type(left) is type(right) and left == right
    if left is None or right is None:
        return left is right
    if isinstance(left, dict) and isinstance(right, dict):
        return left.keys() == right.keys() and all(json_equal(left[k], right[k]) for k in left)
    if isinstance(left, list) and isinstance(right, list):
        return len(left) == len(right) and all(json_equal(a, b) for a, b in zip(left, right))
    if isinstance(left, (int, float)) and isinstance(right, (int, float)):
        return has_type(left, "number") and has_type(right, "number") and left == right
    return type(left) is type(right) and left == right


def has_type(value: Any, kind: str) -> bool:
    kind = TYPE_ALIASES.get(kind, kind)
    if kind == "number":
        return type(value) is int or (type(value) is float and math.isfinite(value))
    if kind == "integer":
        return has_type(value, "number") and (type(value) is int or value.is_integer())
    return {
        "string": isinstance(value, str), "boolean": type(value) is bool,
        "object": isinstance(value, dict), "array": isinstance(value, list),
        "null": value is None,
    }.get(kind, False)


def _resolve_ref(ref: str, root: dict) -> dict:
    if not ref.startswith("#/"):
        raise ValueError("Only local schema references are supported; no URL is fetched")
    result = root
    for part in ref[2:].split("/"):
        result = result[part.replace("~1", "/").replace("~0", "~")]
    return result


def _check_schema_keywords(schema: Any) -> None:
    if isinstance(schema, bool):
        return
    if not isinstance(schema, dict):
        raise ValueError("Invalid bundled schema node")
    unknown = schema.keys() - SCHEMA_KEYWORDS
    if unknown:
        raise ValueError(f"Unsupported bundled schema keywords: {sorted(unknown)}")
    for group in ("properties", "definitions"):
        for child in schema.get(group, {}).values():
            _check_schema_keywords(child)
    for group in ("allOf", "anyOf", "oneOf"):
        for child in schema.get(group, []):
            _check_schema_keywords(child)
    for name in ("items", "not", "if", "then", "else", "additionalProperties"):
        if name in schema:
            _check_schema_keywords(schema[name])


def schema_errors(value: Any, schema: Any, root: dict, path: str = "$") -> list[dict]:
    if isinstance(schema, bool):
        return [] if schema else [issue(path, "schema.forbidden", "Value is forbidden by the bundled schema")]
    errors = []
    if "$ref" in schema:
        # Draft-07: sibling keywords next to $ref do not add constraints.
        return schema_errors(value, _resolve_ref(schema["$ref"], root), root, path)
    if "const" in schema and not json_equal(value, schema["const"]):
        errors.append(issue(path, "schema.const", "Value differs from the required constant"))
    if "enum" in schema and not any(json_equal(value, item) for item in schema["enum"]):
        errors.append(issue(path, "schema.enum", "Value is not in the allowed list"))
    if "type" in schema:
        kinds = schema["type"] if isinstance(schema["type"], list) else [schema["type"]]
        if not any(has_type(value, kind) for kind in kinds):
            errors.append(issue(path, "schema.type", f"Expected type: {', '.join(kinds)}"))
    if isinstance(value, dict):
        for name in schema.get("required", []):
            if name not in value:
                errors.append(issue(f"{path}.{name}", "schema.required", "Required property is missing"))
        properties = schema.get("properties", {})
        for name, child in value.items():
            if name in properties:
                errors.extend(schema_errors(child, properties[name], root, f"{path}.{name}"))
            elif schema.get("additionalProperties") is False:
                errors.append(issue(f"{path}.{name}", "schema.unknown_property", "Unknown property"))
            elif isinstance(schema.get("additionalProperties"), dict):
                errors.extend(schema_errors(child, schema["additionalProperties"], root, f"{path}.{name}"))
    if isinstance(value, list):
        for key, compare in (("minItems", lambda n: len(value) < n), ("maxItems", lambda n: len(value) > n)):
            if key in schema and compare(schema[key]):
                errors.append(issue(path, "schema." + key, f"Array violates {key}"))
        if schema.get("uniqueItems") and any(json_equal(a, b) for i, a in enumerate(value) for b in value[i + 1:]):
            errors.append(issue(path, "schema.uniqueItems", "Array contains duplicates"))
        if "items" in schema:
            for i, child in enumerate(value):
                errors.extend(schema_errors(child, schema["items"], root, f"{path}[{i}]"))
    if isinstance(value, str):
        for key, compare in (("minLength", lambda n: len(value) < n), ("maxLength", lambda n: len(value) > n)):
            if key in schema and compare(schema[key]):
                errors.append(issue(path, "schema." + key, f"String violates {key}"))
    if has_type(value, "number"):
        for key, compare in (("minimum", lambda n: value < n), ("maximum", lambda n: value > n),
                             ("exclusiveMinimum", lambda n: value <= n), ("exclusiveMaximum", lambda n: value >= n)):
            if key in schema and compare(schema[key]):
                errors.append(issue(path, "schema." + key, f"Number violates {key}"))
    for child in schema.get("allOf", []):
        errors.extend(schema_errors(value, child, root, path))
    for group in ("anyOf", "oneOf"):
        if group in schema:
            matches = sum(not schema_errors(value, child, root, path) for child in schema[group])
            if (group == "anyOf" and matches == 0) or (group == "oneOf" and matches != 1):
                errors.append(issue(path, "schema." + group, f"Value violates {group}"))
    if "not" in schema and not schema_errors(value, schema["not"], root, path):
        errors.append(issue(path, "schema.not", "Value matches a forbidden case"))
    if "if" in schema:
        branch = "else" if schema_errors(value, schema["if"], root, path) else "then"
        if branch in schema:
            errors.extend(schema_errors(value, schema[branch], root, path))
    return errors


def _unique_index(items: Any, path: str, errors: list) -> dict:
    if not isinstance(items, list):
        errors.append(issue(path, "catalog.type", "Expected an array"))
        return {}
    index = {}
    for i, item in enumerate(items):
        if not isinstance(item, dict) or not isinstance(item.get("id"), str) or not item["id"]:
            errors.append(issue(f"{path}[{i}]", "catalog.id", "Catalogue item needs a nonempty string id"))
        elif item["id"] in index:
            errors.append(issue(f"{path}[{i}]", "catalog.duplicate_id", "Duplicate catalogue id"))
        else:
            index[item["id"]] = item
    return index


def _package_relative_source(value: Any) -> bool:
    """Lexical check only: never resolve, open, normalize, or fetch this string."""
    if not isinstance(value, str) or not value or value != value.strip():
        return False
    if any(char in value for char in ("\\", ":", "?", "#", "%")) or value.startswith("/"):
        return False
    if any(ord(char) < 32 or ord(char) == 127 for char in value):
        return False
    return all(part not in ("", ".", "..") for part in value.split("/"))


def _validate_sources(sources: list, catalog: dict, errors: list) -> None:
    allowed = catalog.get("allowed_sources")
    if (not isinstance(allowed, list) or not allowed
            or any(not _package_relative_source(value) for value in allowed)):
        errors.append(issue("$catalog.allowed_sources", "catalog.allowed_sources", "A nonempty allowlist of canonical package-relative filenames is required"))
        return
    if len(allowed) != len(set(allowed)):
        errors.append(issue("$catalog.allowed_sources", "catalog.duplicate_source", "Duplicate allowed_sources entry"))
        return
    for i, source in enumerate(sources):
        location = f"$.sources[{i}]"
        if not _package_relative_source(source):
            errors.append(issue(location, "source.path", "Use an exact package-relative filename; URLs, absolute paths, traversal, backslashes and encoded paths are forbidden"))
        elif source not in allowed:
            errors.append(issue(location, "source.not_allowed", "Source is not in catalog.allowed_sources"))


def _field_value(field: dict, spec: dict, path: str, errors: list, warnings: list, strict_signal: bool = False) -> None:
    value = field["value"]
    if value is None:
        return  # Bundled schema checks origin=unknown and note; remains incomplete.
    kind = TYPE_ALIASES.get(spec.get("type"), spec.get("type"))
    if kind not in ("string", "number", "integer", "boolean"):
        errors.append(issue(path, "catalog.field_type", "Catalogue field type is absent or unsupported"))
        return
    if not has_type(value, kind):
        errors.append(issue(path + ".value", "field.type", f"Expected {kind}; booleans are not numbers"))
        return
    if kind in ("number", "integer") and type(value) is int and abs(value) > 9007199254740991:
        errors.append(issue(path + ".value", "field.browser_integer", "Integer exceeds the exact JavaScript safe-integer transport domain; this is a preset format limit, not an inferred server bound"))
        return
    if kind == "string" and not value.strip():
        errors.append(issue(path + ".value", "field.empty", "An empty string is not a completed value; use null with origin=unknown in needs_input"))
        return
    options = spec.get("enum_values")
    if options is not None and not isinstance(options, list):
        errors.append(issue(path, "catalog.enum", "enum_values must be an array or null"))
        return
    if options is not None and not any(json_equal(value, option) for option in options):
        errors.append(issue(path + ".value", "field.enum", "Value is not available in the selected context"))
    if "constant" in spec and not json_equal(value, spec["constant"]):
        errors.append(issue(path + ".value", "field.constant", "Value differs from the educational profile constant"))
    if field["origin"] == "schema_default":
        if "schema_default" not in spec or not json_equal(value, spec["schema_default"]):
            errors.append(issue(path + ".origin", "field.default_origin", "Claimed schema_default does not match the selected context"))
    if strict_signal:
        return  # Signal numeric/conditional domains come only from condition-constraints.json.
    bounds = {key: spec[key] for key in ("minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum") if key in spec}
    numeric_range = spec.get("numeric_range")
    if isinstance(numeric_range, dict):
        for key in ("minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum"):
            if numeric_range.get(key) is not None:
                bounds[key] = numeric_range[key]
    if kind in ("number", "integer"):
        for key, limit in bounds.items():
            if limit is None:
                continue
            if not has_type(limit, "number"):
                errors.append(issue(path, "catalog.bound", f"Invalid numeric constraint: {key}"))
            elif ((key == "minimum" and value < limit) or (key == "maximum" and value > limit)
                  or (key == "exclusiveMinimum" and value <= limit) or (key == "exclusiveMaximum" and value >= limit)):
                errors.append(issue(path + ".value", "field.range", f"Value violates catalogue {key}={limit}"))
        source_constraints = spec.get("source_constraints")
        if isinstance(source_constraints, dict):
            warnings.append(issue(path, "field.source_constraints", "Source-observed numeric constraints are checked; current backend acceptance is not certified"))
            for key, compare in (("minimum", lambda n: value < n), ("maximum", lambda n: value > n)):
                limit = source_constraints.get(key)
                if limit is not None and not has_type(limit, "number"):
                    errors.append(issue(path, "catalog.bound", f"Invalid source constraint: {key}"))
                elif limit is not None and compare(limit):
                    errors.append(issue(path + ".value", "field.source_range", f"Value violates source-observed {key}={limit}; not a runtime-certified range"))
        elif (not any(limit is not None for limit in bounds.values())
              and not field["field_id"].startswith("backtest.")):
            warnings.append(issue(path, "field.range_unresolved", "Numeric bounds are not supplied; type validity does not establish backend acceptance"))


def _macd_order_warning(definition_id: str, values: dict, path: str, warnings: list) -> None:
    """Keep an explicit compatibility warning; numeric and conditional gates use the corpus file."""
    def value(suffix: str) -> Any:
        return values.get(f"{definition_id}.{suffix}", {}).get("value")
    fast, slow = value("fast-period"), value("slow-period")
    if has_type(fast, "number") and has_type(slow, "number") and fast > slow:
        warnings.append(issue(path, "condition.macd_period_order_unverified", "fast-period > slow-period: cross-implementation compatibility is unverified; no exact-transfer conclusion is made"))


def _validate_fields(fields: list, index: dict, path: str, errors: list, warnings: list, required_all: bool = False, strict_signal: bool = False) -> dict:
    found = {}
    for i, field in enumerate(fields):
        field_id = field["field_id"]
        location = f"{path}[{i}]"
        if field_id in found:
            errors.append(issue(location, "field.duplicate_id", "Duplicate field_id in this collection"))
        else:
            found[field_id] = field
        if field_id not in index:
            errors.append(issue(location, "field.unknown_id", "field_id is not defined for this collection"))
        else:
            _field_value(field, index[field_id], location, errors, warnings, strict_signal)
    for field_id, spec in index.items():
        if (required_all or spec.get("required") is True) and field_id not in found:
            errors.append(issue(path, "field.required", f"Missing required field: {field_id}; use null/unknown for an incomplete value"))
    return found


def _read_json(path: Path) -> Any:
    def pairs(values: list) -> dict:
        result = {}
        for key, value in values:
            if key in result:
                raise ValueError(f"Duplicate JSON property: {key}")
            result[key] = value
        return result
    def invalid_constant(value: str) -> None:
        raise ValueError(f"Non-JSON numeric constant: {value}")
    return json.loads(path.read_text(encoding="utf-8-sig"), object_pairs_hook=pairs, parse_constant=invalid_constant)


def _strict_value(value: Any, contract: dict, context_id: str, field_id: str, path: str, errors: list, *, base: bool) -> None:
    """Check one corpus rule without coercion, defaults, downloads or UI-step inference."""
    kind = contract.get("type")
    if not base:
        supported = {"type", "minimum", "maximum", "minimum_inclusive", "maximum_inclusive", "allowed_values", "note", "evidence"}
        if contract.keys() - supported or not any(key in contract for key in ("type", "minimum", "maximum", "allowed_values")):
            errors.append(issue(path, "constraints.rule", "Conditional restriction contains unsupported or absent constraints"))
            return
    if base:
        if contract.get("strict_mode_eligible") is not True or contract.get("status") != "source_observed_and_current_backtest_observed" or kind not in ("integer", "number", "string", "boolean"):
            errors.append(issue(path, "constraints.unresolved", "Selected signal field lacks an eligible, typed corpus contract"))
            return
        if contract.get("nullable") is not False:
            errors.append(issue(path, "constraints.nullability", "Strict signal domain must explicitly disallow null; needs_input is an incomplete placeholder"))
    if value is not None and kind is not None and (not has_type(value, kind) or (kind == "integer" and type(value) is not int)):
        errors.append(issue(path, "constraints.type", "Value does not match the corpus signal type"))
        return
    options = contract.get("allowed_values")
    contextual = contract.get("allowed_values_by_context")
    if base and field_id.endswith((".timeframe", ".signal-action")) and not contextual:
        errors.append(issue(path, "constraints.context", "Timeframe and signal-action domains require an explicit exact-context map"))
        return
    if contextual:
        if not isinstance(contextual, dict) or context_id not in contextual:
            errors.append(issue(path, "constraints.context", "No signal value domain for the exact selected context"))
            return
        options = contextual[context_id]
    if options is not None:
        if not isinstance(options, list) or not options:
            errors.append(issue(path, "constraints.enum_unresolved", "The corpus allowed-value list is malformed or empty"))
        elif value is not None and not any(json_equal(value, option) for option in options):
            errors.append(issue(path, "constraints.allowed_values", "Value is outside the exact corpus allowed-value domain"))
    elif base and kind == "string" and not field_id.endswith(".asset"):
        errors.append(issue(path, "constraints.enum_unresolved", "Non-asset string signal fields need confirmed allowed values"))
    numeric = kind in ("integer", "number")
    for key, inclusive_key in (("minimum", "minimum_inclusive"), ("maximum", "maximum_inclusive")):
        limit = contract.get(key)
        confirmed_unbounded = contract.get("bounds_status") == "source_no_explicit_bounds"
        finite_domain = isinstance(options, list) and bool(options)
        if base and numeric and not has_type(limit, "number") and not (limit is None and (confirmed_unbounded or finite_domain)):
            errors.append(issue(path, "constraints.bounds_unresolved", "Signal numeric minimum and maximum must be confirmed by the corpus"))
            continue
        if limit is None:
            continue
        if not has_type(limit, "number") or type(contract.get(inclusive_key, True)) is not bool:
            errors.append(issue(path, "constraints.bound", "Malformed numeric corpus boundary"))
            continue
        if has_type(value, "number"):
            inclusive = contract.get(inclusive_key, True)
            outside = (value < limit if inclusive else value <= limit) if key == "minimum" else (value > limit if inclusive else value >= limit)
            if outside:
                errors.append(issue(path, "constraints.range", f"Value violates corpus {key}={limit}; this is not a live-backend acceptance claim"))
    if base and numeric:
        step = contract.get("step")
        if not isinstance(step, dict):
            errors.append(issue(path, "constraints.step_unresolved", "Signal numeric step policy is missing"))
            return
        status = step.get("status")
        if status == "server_no_explicit_step":
            if step.get("value") is not None:
                errors.append(issue(path, "constraints.step", "No-explicit-step policy cannot include an invented numeric step"))
        elif status == "source_allowed_values":
            if not isinstance(options, list) or not options or step.get("value") is not None or step.get("anchor") is not None:
                errors.append(issue(path, "constraints.step", "An enumerated numeric domain requires its exact value list and no invented uniform step"))
        elif status in ("integer_normalized_domain", "server_explicit_step", "source_modulo"):
            amount, anchor = step.get("value"), step.get("anchor")
            if not has_type(amount, "number") or amount <= 0 or not has_type(anchor, "number"):
                errors.append(issue(path, "constraints.step", "Confirmed step requires positive value and numeric anchor"))
            elif has_type(value, "number"):
                try:
                    if (Decimal(str(value)) - Decimal(str(anchor))) % Decimal(str(amount)) != 0:
                        errors.append(issue(path, "constraints.step", "Value is not aligned to the confirmed corpus step"))
                except InvalidOperation:
                    errors.append(issue(path, "constraints.step", "Step arithmetic cannot represent this value"))
        elif status == "source_scaled_truncation_modulo":
            if step.get("value") is not None or step.get("anchor") is not None:
                errors.append(issue(path, "constraints.step", "A truncation-based rule is not a uniform decimal step"))
            if not contract.get("numeric_rules"):
                errors.append(issue(path, "constraints.step", "A truncation-based domain requires its exact numeric rule"))
        else:
            errors.append(issue(path, "constraints.step_unresolved", "Unknown numeric step policy cannot authorize a signal value"))
        rules = contract.get("numeric_rules", [])
        if not isinstance(rules, list):
            errors.append(issue(path, "constraints.numeric_rule", "Numeric rules must be an explicit list"))
            return
        for rule in rules:
            if not isinstance(rule, dict) or rule.get("kind") != "scaled_truncation_modulo":
                errors.append(issue(path, "constraints.numeric_rule", "Unsupported numeric rule"))
                continue
            scale, modulus, remainder = rule.get("scale"), rule.get("modulus"), rule.get("remainder")
            if (not has_type(scale, "number") or scale <= 0 or type(modulus) is not int or modulus <= 0
                    or type(remainder) is not int or not 0 <= remainder < modulus):
                errors.append(issue(path, "constraints.numeric_rule", "Invalid scaled-truncation rule"))
            elif has_type(value, "number"):
                try:
                    scaled = float(value) * scale
                    accepted = math.isfinite(scaled) and int(scaled) % modulus == remainder
                except (OverflowError, ValueError):
                    accepted = False
                if not accepted:
                    errors.append(issue(path, "constraints.numeric_rule", "Value violates the source float/truncation/modulo condition"))


def _check_dependency_manifest(conditional: list, cross: list, errors: list) -> None:
    """Pin the complete v0.4 dependency records, independently of the loaded file.

    An empty/truncated rule array is not evidence that there are no dependencies.
    Order is irrelevant; duplicate, missing, altered or unexpected records fail.
    This is a package consistency check, not protection against rewritten code.
    """
    signatures = [sorted(json.dumps(record, sort_keys=True, separators=(",", ":"), allow_nan=False)
                         for record in records) for records in (conditional, cross)]
    digest = hashlib.sha256(json.dumps(signatures, separators=(",", ":")).encode("utf-8")).hexdigest()
    if digest != DEPENDENCY_FINGERPRINT:
        errors.append(issue("$constraints", "constraints.dependency_set",
                            "The complete reviewed dependency set is required, including exact triggers, targets, operators and restrictions"))


def _validate_main_settings(values: dict, errors: list, warnings: list) -> None:
    contract = _read_json(HERE / "main-settings-constraints.json")
    digest = hashlib.sha256(json.dumps(contract, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode("utf-8")).hexdigest()
    if contract.get("format") != "itb.main-settings-constraints/0.5.0" or digest != MAIN_CONSTRAINT_FINGERPRINT:
        errors.append(issue("$main_constraints", "main_constraints.configuration", "The reviewed main-settings contract is required"))
        return
    for field_id, rule in contract["fields"].items():
        value = values.get(field_id, {}).get("value")
        if value is None:
            continue
        if not has_type(value, rule["type"]) or (rule["type"] == "integer" and type(value) is not int):
            errors.append(issue("$.settings", "main_constraints.type", f"{field_id} requires its exact finite JSON type"))
            continue
        gate = rule.get("applies_when")
        if gate:
            selected = values.get(gate["field_id"], {}).get("value")
            if selected is None or json_equal(selected, gate["value"]):
                continue
        for key, inclusive_key in (("minimum", "minimum_inclusive"), ("maximum", "maximum_inclusive")):
            limit, inclusive = rule.get(key), rule.get(inclusive_key, True)
            if limit is None:
                continue
            outside = (value < limit if inclusive else value <= limit) if key == "minimum" else (value > limit if inclusive else value >= limit)
            if outside:
                errors.append(issue("$.settings", "main_constraints.range", f"{field_id} violates source-observed {key}={limit}"))
    multiplier = values.get("backtest.in_martin_gale", {}).get("value")
    if isinstance(multiplier, (int, float)) and not isinstance(multiplier, bool) and multiplier != 1:
        warnings.append(issue("$.settings", "grid.equal_quantities_rejected",
                              "The quantity multiplier differs from 1: if the first-order quantity and every additional-order quantity round to the same value for the symbol quantity step, ordinary BACKTEST rejects the whole deal. Check the quantity step or use a larger first order."))
    warnings.append(issue("$.settings", "main_constraints.runtime_checks", "Main source bounds are checked. Confirm symbol-specific leverage, ticks, order limits, available history and displayed dates in the panel."))


def _validate_symbols(card: dict, values: dict, main_symbol: Any, errors: list, warnings: list) -> None:
    """Check the optional multi-pair list; the tier itself is never asserted, only explained."""
    symbols = card.get("symbols")
    multi = values.get("profile.multi_enabled", {}).get("value")
    if multi is not None and (symbols is not None) != (multi is True):
        errors.append(issue("$.settings", "multi.flag_required",
                            "profile.multi_enabled must be true exactly when the card carries a symbols list"))
    if symbols is None:
        return
    seen: set[str] = set()
    for i, item in enumerate(symbols):
        if not isinstance(item, str) or not re.fullmatch(r"[A-Z0-9_]{1,40}", item):
            errors.append(issue(f"$.symbols[{i}]", "multi.symbol_format",
                                "A pair is an ASCII uppercase ticker of at most 40 characters, as the exchange list spells it"))
            continue
        if item.upper() in seen:
            errors.append(issue(f"$.symbols[{i}]", "multi.symbol_duplicate",
                                "The same pair is listed twice; the tier limit is counted before duplicates are dropped"))
        seen.add(item.upper())
    # A one-entry list is not a multi-pair test but an ordinary card with a stray field. It switches
    # the form into the multi-pair mode, which needs VIP, and the selection cannot be cleared there.
    if len(symbols) == 1:
        errors.append(issue("$.symbols", "multi.single_pair",
                            "A one-entry pair list is an ordinary single-pair card: omit symbols"))
    if len(symbols) > MAX_CARD_SYMBOLS:
        errors.append(issue("$.symbols", "multi.symbol_limit",
                            f"At most {MAX_CARD_SYMBOLS} pairs are transferable; a longer list is refused by the calculation service"))
    if main_symbol is not None and symbols and symbols[0] != main_symbol:
        errors.append(issue("$.symbols", "multi.symbol_mismatch",
                            "The first listed pair must equal backtest.symbol, the card's main pair"))
    if len(symbols or []) > SINGLE_SYMBOL_TIER_LIMIT:
        warnings.append(issue("$.symbols", "multi.vip_required",
                              "A calculation over more than one pair is a VIP feature: the calculation service compares the "
                              f"number of pairs with the account tier limit ({SINGLE_SYMBOL_TIER_LIMIT} pair without VIP) and "
                              "otherwise refuses the run with 400 'Too many symbols'. This package does not observe the "
                              "account tier; say this before offering the link, not after."))
    warnings.append(issue("$.symbols", "multi.selection_unverified",
                          "A pair missing from the exchange list loaded in the form is not ticked and nothing reports it; "
                          "pairs can also be dropped later for missing data. Compare the number of result rows with this list."))
    warnings.append(issue("$.symbols", "multi.independent_runs",
                          "Each pair is calculated separately on the full deposit and the pairs do not interact. This is not "
                          "a portfolio, the summed profit of the result page is not one account's result, and testing more "
                          "pairs produces more results, not better ones."))


def integrate_constraints(card: dict, catalog: dict, constraints: Any) -> dict:
    """Pure, fail-closed signal-domain hook. Call with a structurally valid card."""
    errors: list[dict] = []
    warnings: list[dict] = []
    result = {"errors": errors, "warnings": warnings}
    if not isinstance(constraints, dict) or constraints.get("format") != "itb.condition-constraints/0.4.0" or not isinstance(constraints.get("fields"), dict):
        errors.append(issue("$constraints", "constraints.configuration", "Expected bundled itb.condition-constraints/0.4.0 with explicit field contracts"))
        return result
    contracts = constraints["fields"]
    catalog_fields = {field["id"]: field for definition in catalog["definitions"] for field in definition["fields"]}
    conditional = constraints.get("conditional_rules")
    cross = constraints.get("cross_field_rules")
    if not isinstance(conditional, list) or not isinstance(cross, list):
        errors.append(issue("$constraints", "constraints.configuration", "Explicit conditional_rules and cross_field_rules arrays are required"))
        return result
    _check_dependency_manifest(conditional, cross, errors)
    if errors:
        return result
    for i, condition in enumerate(card["conditions"]):
        definition_id = condition["definition_id"]
        location = f"$.conditions[{i}]"
        values = {field["field_id"]: field["value"] for field in condition["fields"]}
        if definition_id == "impulse.entry" and values.get("impulse.entry.fast-mode") is True:
            errors.append(issue(location, "condition.mode_unverified", "Impulse fm=true is present in metadata, but its ordinary BACKTEST execution behavior is not confirmed in this release. Use a needs-input discussion rather than an automatic preset for that mode."))
        for j, field in enumerate(condition["fields"]):
            field_id, value = field["field_id"], field["value"]
            path = f"{location}.fields[{j}].value"
            contract = contracts.get(field_id)
            if not isinstance(contract, dict) or field_id not in catalog_fields or not field_id.startswith(definition_id + "."):
                errors.append(issue(path, "constraints.unknown_field", "Signal field is not a known corpus field of this exact definition"))
                continue
            if contract.get("type") != catalog_fields[field_id].get("type"):
                errors.append(issue(path, "constraints.catalog_type", "Catalogue and constraint file disagree on the signal type"))
                continue
            if field_id.endswith(".asset"):
                binding = contract.get("binding")
                asset_id = catalog.get("semantic_bindings", {}).get("asset")
                if not isinstance(binding, dict) or binding.get("kind") != "same_as_card_field" or binding.get("field_id") != asset_id:
                    errors.append(issue(path, "constraints.asset_binding", "Signal asset needs an explicit corpus binding to the main asset field"))
            _strict_value(value, contract, card["context_id"], field_id, path, errors, base=True)
        for rule in conditional:
            if not isinstance(rule, dict):
                errors.append(issue("$constraints", "constraints.rule", "Malformed conditional rule"))
                continue
            if rule.get("definition_id") != definition_id:
                continue
            gate, consequent = rule.get("if"), rule.get("then")
            if not isinstance(gate, dict) or set(gate) != {"field_id", "equals"} or gate["field_id"] not in values or not isinstance(consequent, dict) or not isinstance(consequent.get("fields"), dict):
                errors.append(issue(location, "constraints.rule", "Conditional rule has unknown fields or unsupported structure"))
                continue
            for target_id, restriction in consequent["fields"].items():
                if target_id not in values or not isinstance(restriction, dict):
                    errors.append(issue(location, "constraints.rule", "Conditional target is not a selected known field"))
                elif values[gate["field_id"]] is not None and json_equal(values[gate["field_id"]], gate["equals"]):
                    _strict_value(values[target_id], restriction, card["context_id"], target_id, location, errors, base=False)
        operators = {"gt": lambda a, b: a > b, "gte": lambda a, b: a >= b, "lt": lambda a, b: a < b,
                     "lte": lambda a, b: a <= b, "eq": json_equal}
        for rule in cross:
            if not isinstance(rule, dict):
                errors.append(issue("$constraints", "constraints.rule", "Malformed cross-field rule"))
                continue
            if rule.get("definition_id") != definition_id:
                continue
            left, right, operator = rule.get("left_field_id"), rule.get("right_field_id"), rule.get("operator")
            if left not in values or right not in values or operator not in operators:
                errors.append(issue(location, "constraints.rule", "Cross-field rule has unknown fields or operator"))
            elif has_type(values[left], "number") and has_type(values[right], "number") and not operators[operator](values[left], values[right]):
                errors.append(issue(location, "constraints.cross_field", f"Corpus relation {left} {operator} {right} is not satisfied"))
    if card["conditions"]:
        warnings.append(issue("$.conditions", "constraints.source_scope", "Signal domains are checked against the bundled source-observed normalized subset; no deployed parity or execution is certified"))
    return result


def validate_card(card: Any, catalog: Any, schema: dict | None = None, *, constraints: dict | None = None) -> dict:
    """Return {valid, incomplete, errors, warnings}; never authorize execution."""
    errors: list[dict] = []
    warnings: list[dict] = []
    incomplete = isinstance(card, dict) and card.get("status") == "needs_input"
    result = {"valid": False, "incomplete": incomplete, "errors": errors, "warnings": warnings}
    try:
        schema = _read_json(HERE / "strategy-card.schema.json") if schema is None else schema
        _check_schema_keywords(schema)
        errors.extend(schema_errors(card, schema, schema))
    except (ValueError, KeyError, TypeError, OSError) as exc:
        errors.append(issue("$schema", "schema.configuration", str(exc)))
    if errors:
        return result
    if not isinstance(catalog, dict) or catalog.get("format") != "itb.field-catalog/0.5.0":
        errors.append(issue("$catalog", "catalog.format", "Expected itb.field-catalog/0.5.0"))
        return result
    _validate_sources(card["sources"], catalog, errors)
    context_index = _unique_index(catalog.get("contexts"), "$catalog.contexts", errors)
    definitions = _unique_index(catalog.get("definitions"), "$catalog.definitions", errors)
    settings = _unique_index(catalog.get("settings"), "$catalog.settings", errors)
    definition_fields = {}
    seen_catalog_fields = set()
    for definition_id, definition in definitions.items():
        if definition.get("role") not in ROLE_LIMIT_KEYS:
            errors.append(issue("$catalog.definitions", "catalog.role", f"Unknown role: {definition_id}"))
        fields = _unique_index(definition.get("fields"), f"$catalog.definitions.{definition_id}.fields", errors)
        for field_id in fields:
            if field_id in seen_catalog_fields or field_id in settings:
                errors.append(issue("$catalog.definitions", "catalog.duplicate_id", f"Field ID is not globally unique: {field_id}"))
            seen_catalog_fields.add(field_id)
        definition_fields[definition_id] = fields
    if catalog.get("required_definition_fields") is not True:
        errors.append(issue("$catalog.required_definition_fields", "catalog.required_fields", "v0.5 requires every selected definition field"))
    context = context_index.get(card["context_id"])
    if context is None:
        errors.append(issue("$.context_id", "context.unknown_id", "Unknown context_id"))
    if errors:
        return result
    overrides_all = catalog.get("context_overrides", {})
    if not isinstance(overrides_all, dict):
        errors.append(issue("$catalog.context_overrides", "catalog.type", "Expected an object"))
        return result
    for context_id, overrides in overrides_all.items():
        if context_id not in context_index or not isinstance(overrides, dict):
            errors.append(issue("$catalog.context_overrides", "catalog.override", "Unknown context or malformed overrides"))
            continue
        for definition_id, fields in overrides.items():
            if definition_id not in definitions or not isinstance(fields, dict):
                errors.append(issue("$catalog.context_overrides", "catalog.override", "Unknown definition or malformed overrides"))
                continue
            for field_id, replacement in fields.items():
                if field_id not in definition_fields[definition_id] or not isinstance(replacement, dict) or replacement.keys() - {"enum_values", "schema_default", "enum_labels"}:
                    errors.append(issue("$catalog.context_overrides", "catalog.override", "Unknown field or unsupported override properties"))
    if errors:
        return result
    chosen_overrides = overrides_all.get(card["context_id"], {})
    setting_values = _validate_fields(card["settings"], settings, "$.settings", errors, warnings)
    try:
        _validate_main_settings(setting_values, errors, warnings)
    except (OSError, ValueError, KeyError, TypeError) as exc:
        errors.append(issue("$main_constraints", "main_constraints.configuration", str(exc)))
    if card["status"] == "complete_proposal":
        for field_id in USER_REQUIRED_PROPOSAL_SETTINGS:
            field = setting_values.get(field_id)
            if field is None or field.get("origin") != "user" or field.get("value") is None:
                errors.append(issue("$.settings", "proposal.user_input_required", f"User-supplied value is required for {field_id}; AI choices cannot fill this proposal parameter"))
    bindings = {**DEFAULT_BINDINGS, **catalog.get("semantic_bindings", {})}
    def bound(name: str) -> Any:
        return setting_values.get(bindings[name], {}).get("value")
    exchange = bound("exchange")
    if exchange is not None and exchange != context.get("exchange"):
        errors.append(issue("$.settings", "context.exchange", "Exchange differs from the selected context"))
    dca = bound("dca_enabled")
    if type(context.get("dca_enabled")) is not bool:
        errors.append(issue("$catalog.contexts", "catalog.dca", "Context must contain a boolean dca_enabled"))
    elif dca is not None and not json_equal(dca, context["dca_enabled"]):
        errors.append(issue("$.settings", "context.dca", "DCA profile differs from the selected context"))
    mode = bound("mode")
    if mode is not None:
        if mode not in MODE_DCA:
            errors.append(issue("$.settings", "profile.mode", "Mode is outside the v0.5 educational profile"))
        elif not json_equal(MODE_DCA[mode], context.get("dca_enabled")) or (dca is not None and not json_equal(MODE_DCA[mode], dca)):
            errors.append(issue("$.settings", "profile.mode_dca", "DCA must equal the value derived from backtest mode"))
    for name in ("hedge_enabled", "fast_enabled", "optimizer_enabled"):
        profile_id = "profile." + name
        profile_value = setting_values.get(profile_id, {}).get("value")
        if profile_value is not None and profile_value is not False:
            errors.append(issue("$.settings", "profile.unsupported", f"{profile_id} must be false in this educational profile"))
    stop_enabled = setting_values.get("profile.stop_be_enabled", {}).get("value")
    stop_trigger = setting_values.get("backtest.stop_be_trigger_pct")
    stop_level = setting_values.get("backtest.stop_be_val_pct")
    if type(stop_enabled) is bool:
        if stop_enabled:
            if exchange != "BYBIT":
                errors.append(issue("$.settings", "break_even.bybit_only", "Break-even stop is available only in a BYBIT context"))
            if stop_trigger is None or stop_level is None:
                errors.append(issue("$.settings", "break_even.pair_required", "Enabled break-even stop requires both trigger and stop level"))
            else:
                trigger_value, level_value = stop_trigger.get("value"), stop_level.get("value")
                if has_type(trigger_value, "number") and has_type(level_value, "number"):
                    if trigger_value <= 0:
                        errors.append(issue("$.settings", "break_even.trigger", "BE trigger must be greater than zero"))
                    if level_value < 0:
                        errors.append(issue("$.settings", "break_even.level", "BE stop level must be at least zero"))
                    if level_value >= trigger_value:
                        errors.append(issue("$.settings", "break_even.order", "BE stop level must be strictly lower than BE trigger"))
        elif stop_trigger is not None or stop_level is not None:
            errors.append(issue("$.settings", "break_even.disabled_fields", "Disabled break-even stop requires both optional values to be absent"))
    _validate_symbols(card, setting_values, bound("asset"), errors, warnings)
    observation = context.get("schema_observation_id", context.get("observation_id"))
    if observation is not None and card["checks"]["schema_observation_id"] != observation:
        errors.append(issue("$.checks.schema_observation_id", "context.observation", "Schema observation does not match the selected context"))
    role_counts = Counter()
    for i, condition in enumerate(card["conditions"]):
        location = f"$.conditions[{i}]"
        definition_id = condition["definition_id"]
        role_counts[condition["role"]] += 1
        if definition_id not in definitions:
            errors.append(issue(location, "condition.unknown_id", "Unknown definition_id"))
            continue
        if condition["role"] != definitions[definition_id]["role"]:
            errors.append(issue(location + ".role", "condition.role", "Role differs from the selected definition"))
        available = definitions[definition_id].get("available_contexts")
        context_definitions = context.get("available_definition_ids")
        if not isinstance(available, list) or not isinstance(context_definitions, list):
            errors.append(issue(location, "catalog.availability", "Exact definition/context availability is required"))
        elif card["context_id"] not in available or definition_id not in context_definitions:
            errors.append(issue(location, "condition.unavailable", "This definition is unavailable for the selected exchange and mode"))
        field_index = {key: {**spec, **chosen_overrides.get(definition_id, {}).get(key, {})}
                       for key, spec in definition_fields[definition_id].items()}
        values = _validate_fields(condition["fields"], field_index, location + ".fields", errors, warnings, required_all=True, strict_signal=True)
        if definition_id.startswith("macd."):
            _macd_order_warning(definition_id, values, location, warnings)
        for field_id, field in values.items():
            if field_id.endswith(".asset") and field["value"] is not None and bound("asset") is not None and field["value"] != bound("asset"):
                errors.append(issue(location + ".fields", "condition.asset", "Condition asset must equal the main asset"))
            if condition["role"] == "exit" and field_id.endswith(".signal-action") and field["value"] == "AVG" and context.get("dca_enabled") is True:
                warnings.append(issue(location + ".fields", "condition.averaging_action", "This condition requests averaging, not closing."))
    limits = context.get("limits", {})
    if card["status"] in COMPLETE_STATUSES and role_counts["entry"] == 0:
        errors.append(issue("$.conditions", "condition.entry_required", "A completed card requires exactly one signal entry; needs_input may leave entry unspecified"))
    for role, aliases in ROLE_LIMIT_KEYS.items():
        limits_found = [limits[key] for key in aliases if key in limits]
        if not limits_found or any(type(n) is not int or not 0 <= n <= PROFILE_LIMITS[role] for n in limits_found) or len(set(limits_found)) != 1:
            errors.append(issue("$catalog.contexts", "catalog.limit", f"Missing, inconsistent or unsupported {role} limit"))
        elif role_counts[role] > limits_found[0]:
            errors.append(issue("$.conditions", "condition.limit", f"Too many {role} conditions: {role_counts[role]} > {limits_found[0]}"))
    start, end = bound("start"), bound("end")
    dates = []
    for value in (start, end):
        try:
            dates.append(preset_datetime(value) if isinstance(value, str) else None)
        except (ValueError, TypeError):
            dates.append(None)
            errors.append(issue("$.settings", "test.datetime", "Use a real date in YYYY-MM-DDTHH:mm or dd.MM.yyyy HH:mm, without seconds or timezone offsets; no backend timezone is inferred"))
    if all(date is not None for date in dates) and dates[0] > dates[1]:
        errors.append(issue("$.settings", "test.period", "Start must not follow end"))
    try:
        loaded_constraints = _read_json(HERE / "condition-constraints.json") if constraints is None else constraints
        strict_result = integrate_constraints(card, catalog, loaded_constraints)
        errors.extend(strict_result["errors"])
        warnings.extend(strict_result["warnings"])
    except (OSError, ValueError, KeyError, TypeError) as exc:
        errors.append(issue("$constraints", "constraints.configuration", str(exc)))
    if card["status"] in COMPLETE_STATUSES:
        warnings.append(issue("$.status", "card.educational_only", "A completed example or proposal is a filled hypothesis; it is not readiness or permission to execute"))
    result["valid"] = not errors
    return result


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("card", type=Path)
    parser.add_argument("--catalog", type=Path, default=HERE / "field-catalog.json")
    args = parser.parse_args(argv)
    try:
        result = validate_card(_read_json(args.card), _read_json(args.catalog))
    except (OSError, ValueError, TypeError, KeyError) as exc:
        result = {"valid": False, "incomplete": False,
                  "errors": [issue("$input", "input.invalid", str(exc))], "warnings": []}
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 2
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if result["valid"] else 1


if __name__ == "__main__":
    raise SystemExit(main())