1{
2  "op": "replace",
3  "filename": "gplearn/custom_sr.py",
4  "content": "def fitness_function(tree, X, y):\n    \"\"\"Raw MSE fitness used for global best tracking.\"\"\"\n    y_pred = safe_evaluate(tree, X)\n    return float(np.mean((y - y_pred) ** 2))\n\n\n_GP_STATE = {\n    'gen': 0,\n    'best_fitness': float('inf'),\n    'last_improve': 0,\n    'X': None,\n    'y': None,\n    'elite_pred': None,\n    'elite_fitness': None,\n    'parsimony': 0.0,\n    'novelty_weight': 0.01,\n}\n\n\ndef _const_value(node):\n    if node.is_terminal and not (isinstance(node.value, str) and node.value.startswith('x')):\n        try:\n            return float(node.value)\n        except Exception:\n            return None\n    return None\n\n\ndef _make_const(val):\n    return Node(str(round(float(np.clip(val, -1e6, 1e6)), 4)))\n\n\ndef _eval_const(op_name, values):\n    try:\n        arrays = [np.asarray([v], dtype=float) for v in values]\n        out = OPERATORS[op_name][0](*arrays)[0]\n        if not np.isfinite(out):\n            return None\n        return float(np.clip(out, -1e6, 1e6))\n    except Exception:\n        return None\n\n\ndef _simplify(tree):\n    if tree.is_terminal:\n        return tree.copy()\n\n    children = [_simplify(c) for c in tree.children]\n    op = tree.value\n\n    if len(children) == 1:\n        c0 = _const_value(children[0])\n        if c0 is not None:\n            folded = _eval_const(op, [c0])\n            if folded is not None:\n                return _make_const(folded)\n        return Node(op, children)\n\n    a, b = children\n    ca, cb = _const_value(a), _const_value(b)\n\n    if op == 'add':\n        if ca is not None and abs(ca) < 1e-12:\n            return b.copy()\n        if cb is not None and abs(cb) < 1e-12:\n            return a.copy()\n    elif op == 'sub':\n        if cb is not None and abs(cb) < 1e-12:\n            return a.copy()\n        if str(a) == str(b):\n            return _make_const(0.0)\n    elif op == 'mul':\n        if (ca is not None and abs(ca) < 1e-12) or (cb is not None and abs(cb) < 1e-12):\n            return _make_const(0.0)\n        if ca is not None and abs(ca - 1.0) < 1e-12:\n            return b.copy()\n        if cb is not None and abs(cb - 1.0) < 1e-12:\n            return a.copy()\n    elif op == 'div':\n        if ca is not None and abs(ca) < 1e-12:\n            return _make_const(0.0)\n        if cb is not None and abs(cb - 1.0) < 1e-12:\n            return a.copy()\n        if str(a) == str(b):\n            return _make_const(1.0)\n\n    if ca is not None and cb is not None:\n        folded = _eval_const(op, [ca, cb])\n        if folded is not None:\n            return _make_const(folded)\n\n    return Node(op, children)\n\n\ndef _selection_score(idx, fitnesses, sizes, preds=None):\n    score = float(fitnesses[idx])\n    elite_pred = _GP_STATE.get('elite_pred')\n    y = _GP_STATE.get('y')\n    if elite_pred is not None and preds is not None and y is not None:\n        scale = float(np.mean(np.abs(y)) + 1e-6)\n        novelty = float(np.mean(np.abs(preds[idx] - elite_pred))) / scale\n        score -= _GP_STATE.get('novelty_weight', 0.01) * novelty / (1.0 + score)\n    return (score, float(sizes[idx]) if sizes is not None else 0.0)\n\n\ndef _tree_score(tree):\n    X = _GP_STATE.get('X')\n    y = _GP_STATE.get('y')\n    if X is None or y is None:\n        return float(tree.size())\n    pred = safe_evaluate(tree, X)\n    mse = float(np.mean((y - pred) ** 2))\n    size_pen = 0.5 * _GP_STATE.get('parsimony', 0.0) * tree.size()\n    elite_pred = _GP_STATE.get('elite_pred')\n    novelty_bonus = 0.0\n    if elite_pred is not None:\n        scale = float(np.mean(np.abs(y)) + 1e-6)\n        novelty = float(np.mean(np.abs(pred - elite_pred))) / scale\n        novelty_bonus = _GP_STATE.get('novelty_weight', 0.01) * novelty / (1.0 + mse)\n    return mse + size_pen - novelty_bonus\n\n\ndef selection(population, fitnesses, n_select, _errors=None, _preds=None,\n              _sizes=None, _case_weights=None):\n    \"\"\"Residual-weighted epsilon lexicase on small candidate pools.\"\"\"\n    selected = []\n    pop_size = len(population)\n\n    for _ in range(n_select):\n        if _errors is None:\n            candidates = random.sample(range(pop_size), min(7, pop_size))\n        else:\n            candidates = random.sample(range(pop_size), min(32, pop_size))\n            n_cases = _errors.shape[1]\n            if _case_weights is None:\n                cases = list(range(n_cases))\n                random.shuffle(cases)\n            else:\n                cases = list(np.random.choice(\n                    n_cases, size=n_cases, replace=False, p=_case_weights\n                ))\n\n            for case in cases:\n                if len(candidates) <= 2:\n                    break\n                errs = _errors[candidates, case]\n                best = float(np.min(errs))\n                med = float(np.median(errs))\n                mad = float(np.median(np.abs(errs - med)))\n                eps = best + 0.5 * mad + 0.05 * (med - best)\n                candidates = [i for i, e in zip(candidates, errs) if e <= eps + 1e-12]\n\n        winner = min(candidates, key=lambda i: _selection_score(i, fitnesses, _sizes, _preds))\n        selected.append(population[winner].copy())\n    return selected\n\n\ndef crossover(parent1, parent2, n_features, max_depth=17):\n    \"\"\"Size-fair subtree crossover with semantic lookahead.\"\"\"\n    best = parent1.copy()\n    best_score = _tree_score(best)\n    donor_nodes_full = parent2.get_all_nodes()\n    if len(parent1.get_all_nodes()) <= 1 or len(donor_nodes_full) == 0:\n        return best\n\n    for _ in range(4):\n        offspring = parent1.copy()\n        off_nodes = offspring.get_all_nodes()\n        if len(off_nodes) <= 1:\n            break\n\n        off_point = random.randint(1, len(off_nodes) - 1)\n        target_size = off_nodes[off_point][0].size()\n        donor_pool = random.sample(donor_nodes_full, min(6, len(donor_nodes_full)))\n        donor_subtree = min(donor_pool, key=lambda t: abs(t[0].size() - target_size))[0].copy()\n\n        _, parent, child_idx = off_nodes[off_point]\n        if parent is not None:\n            parent.children[child_idx] = donor_subtree\n        else:\n            offspring = donor_subtree\n\n        offspring = _simplify(offspring)\n        if offspring.depth() > max_depth:\n            continue\n\n        score = _tree_score(offspring)\n        if score < best_score:\n            best, best_score = offspring, score\n\n    return best\n\n\ndef mutation(parent, n_features, max_depth=17):\n    \"\"\"Multi-operator mutation with shrink and constant refinement.\"\"\"\n    best = parent.copy()\n    best_score = _tree_score(best)\n\n    for kind in ['subtree', 'point', 'shrink', 'const', 'subtree']:\n        offspring = parent.copy()\n        nodes = offspring.get_all_nodes()\n\n        if kind == 'subtree':\n            idx = 0 if len(nodes) == 1 else random.randint(1, len(nodes) - 1)\n            new_subtree = generate_tree('grow', 3, n_features)\n            _, par, child_idx = nodes[idx]\n            if par is None:\n                offspring = new_subtree\n            else:\n                par.children[child_idx] = new_subtree\n\n        elif kind == 'point':\n            op_nodes = [(i, n) for i, (n, _, _) in enumerate(nodes) if not n.is_terminal]\n            if not op_nodes:\n                continue\n            idx, node = random.choice(op_nodes)\n            arity = len(node.children)\n            alternatives = [op for op, (_, a) in OPERATORS.items() if a == arity and op != node.value]\n            if not alternatives:\n                continue\n            node.value = random.choice(alternatives)\n\n        elif kind == 'shrink':\n            internal = [(i, n, p, c) for i, (n, p, c) in enumerate(nodes) if len(n.children) > 0]\n            if not internal:\n                continue\n            _, node, par, child_idx = random.choice(internal)\n            replacement = random.choice(node.children).copy()\n            if par is None:\n                offspring = replacement\n            else:\n                par.children[child_idx] = replacement\n\n        else:  # const\n            numeric = [(i, n) for i, (n, _, _) in enumerate(nodes)\n                       if n.is_terminal and _const_value(n) is not None]\n            if numeric:\n                _, node = random.choice(numeric)\n                c = float(node.value)\n                node.value = str(round(c + random.gauss(0.0, random.choice([1.0, 0.5, 0.1])), 4))\n            else:\n                terminals = [(i, n) for i, (n, _, _) in enumerate(nodes) if n.is_terminal]\n                if not terminals:\n                    continue\n                _, node = random.choice(terminals)\n                if isinstance(node.value, str) and node.value.startswith('x') and n_features > 1:\n                    node.value = f'x{random.randint(0, n_features - 1)}'\n                else:\n                    node.value = str(round(random.uniform(-3.0, 3.0), 4))\n\n        offspring = _simplify(offspring)\n        if offspring.depth() > max_depth:\n            continue\n\n        score = _tree_score(offspring)\n        if score < best_score:\n            best, best_score = offspring, score\n\n    return best\n\n\ndef evolve_one_generation(population, fitnesses, X_train, y_train,\n                          n_features, pop_size,\n                          crossover_rate=0.9, mutation_rate=0.05,\n                          max_depth=17):\n    \"\"\"Residual-guided lexicase GP with lookahead variation and dual elitism.\"\"\"\n    raw_fit = np.asarray(fitnesses, dtype=float)\n    sizes = np.asarray([tree.size() for tree in population], dtype=float)\n    preds = np.asarray([safe_evaluate(tree, X_train) for tree in population])\n    errors = np.abs(preds - y_train[None, :])\n\n    elite_idx = int(np.argmin(raw_fit))\n    best_fit = float(raw_fit[elite_idx])\n    _GP_STATE['gen'] += 1\n    if best_fit + 1e-12 < _GP_STATE.get('best_fitness', float('inf')):\n        _GP_STATE['best_fitness'] = best_fit\n        _GP_STATE['last_improve'] = _GP_STATE['gen']\n    stagnation = _GP_STATE['gen'] - _GP_STATE.get('last_improve', 0)\n\n    len_var = float(np.var(sizes))\n    if len_var > 1e-15:\n        parsimony = float(np.cov(s\n... (truncated)",
5  "after_line": 306,
6  "start_line": 228,
7  "end_line": 306
8}

1{
2  "op": "replace",
3  "filename": "gplearn/custom_sr.py",
4  "content": "def fitness_function(tree, X, y):\n    \"\"\"Raw MSE fitness used for global best tracking.\"\"\"\n    y_pred = safe_evaluate(tree, X)\n    return float(np.mean((y - y_pred) ** 2))\n\n\n_GP_STATE = {\n    'gen': 0,\n    'best_fitness': float('inf'),\n    'last_improve': 0,\n    'X': None,\n    'y': None,\n    'elite_pred': None,\n    'elite_fitness': None,\n    'parsimony': 0.0,\n    'novelty_weight': 0.01,\n}\n\n\ndef _const_value(node):\n    if node.is_terminal and not (isinstance(node.value, str) and node.value.startswith('x')):\n        try:\n            return float(node.value)\n        except Exception:\n            return None\n    return None\n\n\ndef _make_const(val):\n    return Node(str(round(float(np.clip(val, -1e6, 1e6)), 4)))\n\n\ndef _eval_const(op_name, values):\n    try:\n        arrays = [np.asarray([v], dtype=float) for v in values]\n        out = OPERATORS[op_name][0](*arrays)[0]\n        if not np.isfinite(out):\n            return None\n        return float(np.clip(out, -1e6, 1e6))\n    except Exception:\n        return None\n\n\ndef _scaled_mse_from_pred(pred, y):\n    pm = float(np.mean(pred))\n    ym = float(np.mean(y))\n    centered_pred = pred - pm\n    denom = float(np.dot(centered_pred, centered_pred))\n    if denom < 1e-12:\n        scaled = np.full_like(pred, ym)\n    else:\n        a = float(np.dot(centered_pred, y - ym) / denom)\n        b = ym - a * pm\n        scaled = a * pred + b\n    return float(np.mean((y - scaled) ** 2))\n\n\ndef _simplify(tree):\n    if tree.is_terminal:\n        return tree.copy()\n\n    children = [_simplify(c) for c in tree.children]\n    op = tree.value\n\n    if len(children) == 1:\n        c0 = _const_value(children[0])\n        if c0 is not None:\n            folded = _eval_const(op, [c0])\n            if folded is not None:\n                return _make_const(folded)\n        return Node(op, children)\n\n    a, b = children\n    ca, cb = _const_value(a), _const_value(b)\n\n    if op == 'add':\n        if ca is not None and abs(ca) < 1e-12:\n            return b.copy()\n        if cb is not None and abs(cb) < 1e-12:\n            return a.copy()\n    elif op == 'sub':\n        if cb is not None and abs(cb) < 1e-12:\n            return a.copy()\n        if str(a) == str(b):\n            return _make_const(0.0)\n    elif op == 'mul':\n        if (ca is not None and abs(ca) < 1e-12) or (cb is not None and abs(cb) < 1e-12):\n            return _make_const(0.0)\n        if ca is not None and abs(ca - 1.0) < 1e-12:\n            return b.copy()\n        if cb is not None and abs(cb - 1.0) < 1e-12:\n            return a.copy()\n    elif op == 'div':\n        if ca is not None and abs(ca) < 1e-12:\n            return _make_const(0.0)\n        if cb is not None and abs(cb - 1.0) < 1e-12:\n            return a.copy()\n        if str(a) == str(b):\n            return _make_const(1.0)\n\n    if ca is not None and cb is not None:\n        folded = _eval_const(op, [ca, cb])\n        if folded is not None:\n            return _make_const(folded)\n\n    return Node(op, children)\n\n\ndef _selection_score(idx, raw_fitnesses, sizes, preds=None):\n    score = float(raw_fitnesses[idx])\n    y = _GP_STATE.get('y')\n    if preds is not None and y is not None:\n        scaled_mse = _scaled_mse_from_pred(preds[idx], y)\n        score = 0.7 * score + 0.3 * scaled_mse\n        elite_pred = _GP_STATE.get('elite_pred')\n        if elite_pred is not None:\n            scale = float(np.mean(np.abs(y)) + 1e-6)\n            novelty = float(np.mean(np.abs(preds[idx] - elite_pred))) / scale\n            score -= _GP_STATE.get('novelty_weight', 0.01) * novelty / (1.0 + score)\n    score += _GP_STATE.get('parsimony', 0.0) * (float(sizes[idx]) if sizes is not None else 0.0)\n    return (score, float(sizes[idx]) if sizes is not None else 0.0)\n\n\ndef _tree_score(tree):\n    X = _GP_STATE.get('X')\n    y = _GP_STATE.get('y')\n    if X is None or y is None:\n        return float(tree.size())\n    pred = safe_evaluate(tree, X)\n    mse = float(np.mean((y - pred) ** 2))\n    scaled_mse = _scaled_mse_from_pred(pred, y)\n    score = 0.8 * mse + 0.2 * scaled_mse\n    score += 0.5 * _GP_STATE.get('parsimony', 0.0) * tree.size()\n    elite_pred = _GP_STATE.get('elite_pred')\n    if elite_pred is not None:\n        scale = float(np.mean(np.abs(y)) + 1e-6)\n        novelty = float(np.mean(np.abs(pred - elite_pred))) / scale\n        score -= _GP_STATE.get('novelty_weight', 0.01) * novelty / (1.0 + score)\n    return score\n\n\ndef selection(population, fitnesses, n_select, _errors=None, _preds=None,\n              _sizes=None, _case_weights=None):\n    \"\"\"Residual-weighted epsilon lexicase on small candidate pools.\"\"\"\n    selected = []\n    pop_size = len(population)\n\n    for _ in range(n_select):\n        if _errors is None:\n            candidates = random.sample(range(pop_size), min(7, pop_size))\n        else:\n            candidates = random.sample(range(pop_size), min(32, pop_size))\n            n_cases = _errors.shape[1]\n            if _case_weights is None:\n                cases = list(range(n_cases))\n                random.shuffle(cases)\n            else:\n                cases = list(np.random.choice(\n                    n_cases, size=n_cases, replace=False, p=_case_weights\n                ))\n\n            for case in cases:\n                if len(candidates) <= 2:\n                    break\n                errs = _errors[candidates, case]\n                best = float(np.min(errs))\n                med = float(np.median(errs))\n                mad = float(np.median(np.abs(errs - med)))\n                eps = best + 0.5 * mad + 0.05 * (med - best)\n                candidates = [i for i, e in zip(candidates, errs) if e <= eps + 1e-12]\n\n        winner = min(candidates, key=lambda i: _selection_score(i, fitnesses, _sizes, _preds))\n        selected.append(population[winner].copy())\n    return selected\n\n\ndef crossover(parent1, parent2, n_features, max_depth=17):\n    \"\"\"Size-fair subtree crossover with semantic lookahead.\"\"\"\n    best = parent1.copy()\n    best_score = _tree_score(best)\n    donor_nodes_full = parent2.get_all_nodes()\n    if len(parent1.get_all_nodes()) <= 1 or len(donor_nodes_full) == 0:\n        return best\n\n    for _ in range(4):\n        offspring = parent1.copy()\n        off_nodes = offspring.get_all_nodes()\n        if len(off_nodes) <= 1:\n            break\n\n        off_point = random.randint(1, len(off_nodes) - 1)\n        target_size = off_nodes[off_point][0].size()\n        donor_pool = random.sample(donor_nodes_full, min(6, len(donor_nodes_full)))\n        donor_subtree = min(donor_pool, key=lambda t: abs(t[0].size() - target_size))[0].copy()\n\n        _, parent, child_idx = off_nodes[off_point]\n        if parent is not None:\n            parent.children[child_idx] = donor_subtree\n        else:\n            offspring = donor_subtree\n\n        offspring = _simplify(offspring)\n        if offspring.depth() > max_depth:\n            continue\n\n        score = _tree_score(offspring)\n        if score < best_score:\n            best, best_score = offspring, score\n\n    return best\n\n\ndef mutation(parent, n_features, max_depth=17):\n    \"\"\"Multi-operator mutation with shrink and constant refinement.\"\"\"\n    best = parent.copy()\n    best_score = _tree_score(best)\n\n    for kind in ['subtree', 'point', 'shrink', 'const', 'subtree']:\n        offspring = parent.copy()\n        nodes = offspring.get_all_nodes()\n\n        if kind == 'subtree':\n            idx = 0 if len(nodes) == 1 else random.randint(1, len(nodes) - 1)\n            new_subtree = generate_tree('grow', 3, n_features)\n            _, par, child_idx = nodes[idx]\n            if par is None:\n                offspring = new_subtree\n            else:\n                par.children[child_idx] = new_subtree\n\n        elif kind == 'point':\n            op_nodes = [(i, n) for i, (n, _, _) in enumerate(nodes) if not n.is_terminal]\n            if not op_nodes:\n                continue\n            _, node = random.choice(op_nodes)\n            arity = len(node.children)\n            alternatives = [op for op, (_, a) in OPERATORS.items() if a == arity and op != node.value]\n            if not alternatives:\n                continue\n            node.value = random.choice(alternatives)\n\n        elif kind == 'shrink':\n            internal = [(n, p, c) for n, p, c in [(n, p, c) for n, p, c in nodes] if len(n.children) > 0]\n            if not internal:\n                continue\n            node, par, child_idx = random.choice(internal)\n            replacement = random.choice(node.children).copy()\n            if par is None:\n                offspring = replacement\n            else:\n                par.children[child_idx] = replacement\n\n        else:  # const\n            numeric = [n for n, _, _ in nodes if n.is_terminal and _const_value(n) is not None]\n            if numeric:\n                node = random.choice(numeric)\n                base = float(node.value)\n                candidates = [base + random.gauss(0.0, s) for s in (1.0, 0.5, 0.1)]\n                node.value = str(round(random.choice(candidates), 4))\n            else:\n                terminals = [n for n, _, _ in nodes if n.is_terminal]\n                if not terminals:\n                    continue\n                node = random.choice(terminals)\n                if isinstance(node.value, str) and node.value.startswith('x') and n_features > 1:\n                    node.value = f'x{random.randint(0, n_features - 1)}'\n                else:\n                    node.value = str(round(random.uniform(-3.0, 3.0), 4))\n\n        offspring = _simplify(offspring)\n        if offspring.depth() > max_depth:\n            continue\n\n        score = _tree_score(offspring)\n        if score < best_score:\n            best, best_score = offspring, score\n\n    return best\n\n\ndef evolve_one_generation(population, fitnesses, X_train, y_train,\n                          n_features, pop_size,\n                          crossover_rate=0.9, mutation_rate=0.05,\n                          max_depth=17):\n    \"\"\"Residual-guided lexicase GP with lookahead variation and dual elitism.\"\"\"\n    raw_fit = np.asa\n... (truncated)",
5  "after_line": 554,
6  "start_line": 228,
7  "end_line": 554
8}

1{
2  "n": 3
3}