| 1496 | return size |
| 1497 | |
| 1498 | def _optimize_grid() -> Union[pd.Series, Tuple[pd.Series, pd.Series]]: |
| 1499 | rand = default_rng(random_state).random |
| 1500 | grid_frac = (1 if max_tries is None else |
| 1501 | max_tries if 0 < max_tries <= 1 else |
| 1502 | max_tries / _grid_size()) |
| 1503 | param_combos = [dict(params) # back to dict so it pickles |
| 1504 | for params in (AttrDict(params) |
| 1505 | for params in product(*(zip(repeat(k), _tuple(v)) |
| 1506 | for k, v in kwargs.items()))) |
| 1507 | if constraint(params) |
| 1508 | and rand() <= grid_frac] |
| 1509 | if not param_combos: |
| 1510 | raise ValueError('No admissible parameter combinations to test') |
| 1511 | |
| 1512 | if len(param_combos) > 300: |
| 1513 | warnings.warn(f'Searching for best of {len(param_combos)} configurations.', |
| 1514 | stacklevel=2) |
| 1515 | |
| 1516 | heatmap = pd.Series(np.nan, |
| 1517 | name=maximize_key, |
| 1518 | index=pd.MultiIndex.from_tuples( |
| 1519 | [p.values() for p in param_combos], |
| 1520 | names=next(iter(param_combos)).keys())) |
| 1521 | |
| 1522 | from . import Pool |
| 1523 | with Pool() as pool, \ |
| 1524 | SharedMemoryManager() as smm: |
| 1525 | with patch(self, '_data', None): |
| 1526 | bt = copy(self) # bt._data will be reassigned in _mp_task worker |
| 1527 | results = _tqdm( |
| 1528 | pool.imap(Backtest._mp_task, |
| 1529 | ((bt, smm.df2shm(self._data), params_batch) |
| 1530 | for params_batch in _batch(param_combos))), |
| 1531 | total=len(param_combos), |
| 1532 | desc='Backtest.optimize' |
| 1533 | ) |
| 1534 | for param_batch, result in zip(_batch(param_combos), results): |
| 1535 | for params, stats in zip(param_batch, result): |
| 1536 | if stats is not None: |
| 1537 | heatmap[tuple(params.values())] = maximize(stats) |
| 1538 | |
| 1539 | if pd.isnull(heatmap).all(): |
| 1540 | # No trade was made in any of the runs. Just make a random |
| 1541 | # run so we get some, if empty, results |
| 1542 | stats = self.run(**param_combos[0]) |
| 1543 | else: |
| 1544 | best_params = heatmap.idxmax(skipna=True) |
| 1545 | stats = self.run(**dict(zip(heatmap.index.names, best_params))) |
| 1546 | |
| 1547 | if return_heatmap: |
| 1548 | return stats, heatmap |
| 1549 | return stats |
| 1550 | |
| 1551 | def _optimize_sambo() -> Union[pd.Series, |
| 1552 | Tuple[pd.Series, pd.Series], |