forked from pymc-devs/pymc
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathsampling.py
1554 lines (1352 loc) · 59.7 KB
/
sampling.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
from typing import Dict, List, Optional, TYPE_CHECKING, cast
if TYPE_CHECKING:
from typing import Any
from typing import Iterable as TIterable
from collections import defaultdict, Iterable
from copy import copy
import pickle
import logging
import warnings
import numpy as np
import theano.gradient as tg
from theano.tensor import Tensor
from .backends.base import BaseTrace, MultiTrace
from .backends.ndarray import NDArray
from .distributions.distribution import draw_values
from .model import modelcontext, Point, all_continuous, Model
from .step_methods import (NUTS, HamiltonianMC, Metropolis, BinaryMetropolis,
BinaryGibbsMetropolis, CategoricalGibbsMetropolis,
Slice, CompoundStep, arraystep, smc)
from .util import update_start_vals, get_untransformed_name, is_transformed_name, get_default_varnames
from .vartypes import discrete_types
from .exceptions import IncorrectArgumentsError
from pymc3.step_methods.hmc import quadpotential
import pymc3 as pm
from tqdm import tqdm
import sys
sys.setrecursionlimit(10000)
__all__ = ['sample', 'iter_sample', 'sample_posterior_predictive',
'sample_posterior_predictive_w', 'init_nuts',
'sample_prior_predictive', 'sample_ppc', 'sample_ppc_w']
STEP_METHODS = (NUTS, HamiltonianMC, Metropolis, BinaryMetropolis,
BinaryGibbsMetropolis, Slice, CategoricalGibbsMetropolis)
_log = logging.getLogger('pymc3')
def instantiate_steppers(model, steps, selected_steps, step_kwargs=None):
"""Instantiates steppers assigned to the model variables.
This function is intended to be called automatically from `sample()`, but
may be called manually.
Parameters
----------
model : Model object
A fully-specified model object
step : step function or vector of step functions
One or more step functions that have been assigned to some subset of
the model's parameters. Defaults to None (no assigned variables).
selected_steps: dictionary of step methods and variables
The step methods and the variables that have were assigned to them.
step_kwargs : dict
Parameters for the samplers. Keys are the lower case names of
the step method, values a dict of arguments.
Returns
-------
methods : list
List of step methods associated with the model's variables.
"""
if step_kwargs is None:
step_kwargs = {}
used_keys = set()
for step_class, vars in selected_steps.items():
if len(vars) == 0:
continue
args = step_kwargs.get(step_class.name, {})
used_keys.add(step_class.name)
step = step_class(vars=vars, **args)
steps.append(step)
unused_args = set(step_kwargs).difference(used_keys)
if unused_args:
raise ValueError('Unused step method arguments: %s' % unused_args)
if len(steps) == 1:
steps = steps[0]
return steps
def assign_step_methods(model, step=None, methods=STEP_METHODS,
step_kwargs=None):
"""Assign model variables to appropriate step methods.
Passing a specified model will auto-assign its constituent stochastic
variables to step methods based on the characteristics of the variables.
This function is intended to be called automatically from `sample()`, but
may be called manually. Each step method passed should have a
`competence()` method that returns an ordinal competence value
corresponding to the variable passed to it. This value quantifies the
appropriateness of the step method for sampling the variable.
Parameters
----------
model : Model object
A fully-specified model object
step : step function or vector of step functions
One or more step functions that have been assigned to some subset of
the model's parameters. Defaults to None (no assigned variables).
methods : vector of step method classes
The set of step methods from which the function may choose. Defaults
to the main step methods provided by PyMC3.
step_kwargs : dict
Parameters for the samplers. Keys are the lower case names of
the step method, values a dict of arguments.
Returns
-------
methods : list
List of step methods associated with the model's variables.
"""
steps = []
assigned_vars = set()
if step is not None:
try:
steps += list(step)
except TypeError:
steps.append(step)
for step in steps:
try:
assigned_vars = assigned_vars.union(set(step.vars))
except AttributeError:
for method in step.methods:
assigned_vars = assigned_vars.union(set(method.vars))
# Use competence classmethods to select step methods for remaining
# variables
selected_steps = defaultdict(list)
for var in model.free_RVs:
if var not in assigned_vars:
# determine if a gradient can be computed
has_gradient = var.dtype not in discrete_types
if has_gradient:
try:
tg.grad(model.logpt, var)
except (AttributeError,
NotImplementedError,
tg.NullTypeGradError):
has_gradient = False
# select the best method
selected = max(methods, key=lambda method,
var=var, has_gradient=has_gradient:
method._competence(var, has_gradient))
selected_steps[selected].append(var)
return instantiate_steppers(model, steps, selected_steps, step_kwargs)
def _print_step_hierarchy(s, level=0):
if isinstance(s, (list, tuple)):
_log.info('>' * level + 'list')
for i in s:
_print_step_hierarchy(i, level+1)
elif isinstance(s, CompoundStep):
_log.info('>' * level + 'CompoundStep')
for i in s.methods:
_print_step_hierarchy(i, level+1)
else:
varnames = ', '.join([get_untransformed_name(v.name) if is_transformed_name(v.name)
else v.name for v in s.vars])
_log.info('>' * level + '{}: [{}]'.format(s.__class__.__name__, varnames))
def _cpu_count():
"""Try to guess the number of CPUs in the system.
We use the number provided by psutil if that is installed.
If not, we use the number provided by multiprocessing, but assume
that half of the cpus are only hardware threads and ignore those.
"""
try:
import psutil
cpus = psutil.cpu_count(False)
except ImportError:
import multiprocessing
try:
cpus = multiprocessing.cpu_count() // 2
except NotImplementedError:
cpus = 1
if cpus is None:
cpus = 1
return cpus
def sample(draws=500, step=None, init='auto', n_init=200000, start=None, trace=None, chain_idx=0,
chains=None, cores=None, tune=500, progressbar=True,
model=None, random_seed=None, discard_tuned_samples=True,
compute_convergence_checks=True, **kwargs):
"""Draw samples from the posterior using the given step methods.
Multiple step methods are supported via compound step methods.
Parameters
----------
draws : int
The number of samples to draw. Defaults to 500. The number of tuned samples are discarded
by default. See discard_tuned_samples.
step : function or iterable of functions
A step function or collection of functions. If there are variables without a step methods,
step methods for those variables will be assigned automatically.
init : str
Initialization method to use for auto-assigned NUTS samplers.
* auto : Choose a default initialization method automatically.
Currently, this is `'jitter+adapt_diag'`, but this can change in the future.
If you depend on the exact behaviour, choose an initialization method explicitly.
* adapt_diag : Start with a identity mass matrix and then adapt a diagonal based on the
variance of the tuning samples. All chains use the test value (usually the prior mean)
as starting point.
* jitter+adapt_diag : Same as `adapt_diag`, but add uniform jitter in [-1, 1] to the
starting point in each chain.
* advi+adapt_diag : Run ADVI and then adapt the resulting diagonal mass matrix based on the
sample variance of the tuning samples.
* advi+adapt_diag_grad : Run ADVI and then adapt the resulting diagonal mass matrix based
on the variance of the gradients during tuning. This is **experimental** and might be
removed in a future release.
* advi : Run ADVI to estimate posterior mean and diagonal mass matrix.
* advi_map: Initialize ADVI with MAP and use MAP as starting point.
* map : Use the MAP as starting point. This is discouraged.
* nuts : Run NUTS and estimate posterior mean and mass matrix from the trace.
n_init : int
Number of iterations of initializer. Only works for 'nuts' and 'ADVI'.
If 'ADVI', number of iterations, if 'nuts', number of draws.
start : dict, or array of dict
Starting point in parameter space (or partial point)
Defaults to trace.point(-1)) if there is a trace provided and model.test_point if not
(defaults to empty dict). Initialization methods for NUTS (see `init` keyword) can
overwrite the default. For 'SMC' it should be a list of dict with length `chains`.
trace : backend, list, or MultiTrace
This should be a backend instance, a list of variables to track, or a MultiTrace object
with past values. If a MultiTrace object is given, it must contain samples for the chain
number `chain`. If None or a list of variables, the NDArray backend is used.
Passing either "text" or "sqlite" is taken as a shortcut to set up the corresponding
backend (with "mcmc" used as the base name). Ignored when using 'SMC'.
chain_idx : int
Chain number used to store sample in backend. If `chains` is greater than one, chain
numbers will start here. Ignored when using 'SMC'.
chains : int
The number of chains to sample. Running independent chains is important for some
convergence statistics and can also reveal multiple modes in the posterior. If `None`,
then set to either `cores` or 2, whichever is larger. For SMC the number of chains is the
number of draws.
cores : int
The number of chains to run in parallel. If `None`, set to the number of CPUs in the
system, but at most 4 (for 'SMC' ignored if `pm.SMC(parallel=False)`. Keep in mind that
some chains might themselves be multithreaded via openmp or BLAS. In those cases it might
be faster to set this to 1.
tune : int
Number of iterations to tune, defaults to 500. Ignored when using 'SMC'. Samplers adjust
the step sizes, scalings or similar during tuning. Tuning samples will be drawn in addition
to the number specified in the `draws` argument, and will be discarded unless
`discard_tuned_samples` is set to False.
progressbar : bool
Whether or not to display a progress bar in the command line. The bar shows the percentage
of completion, the sampling speed in samples per second (SPS), and the estimated remaining
time until completion ("expected time of arrival"; ETA).
model : Model (optional if in `with` context)
random_seed : int or list of ints
A list is accepted if `cores` is greater than one.
discard_tuned_samples : bool
Whether to discard posterior samples of the tune interval. Ignored when using 'SMC'
compute_convergence_checks : bool, default=True
Whether to compute sampler statistics like gelman-rubin and effective_n.
Ignored when using 'SMC'
Returns
-------
trace : pymc3.backends.base.MultiTrace
A `MultiTrace` object that contains the samples.
Notes
-----
Optional keyword arguments can be passed to `sample` to be delivered to the
`step_method`s used during sampling. In particular, the NUTS step method accepts
a number of arguments. Common options are:
* target_accept: float in [0, 1]. The step size is tuned such that we approximate this
acceptance rate. Higher values like 0.9 or 0.95 often work better for problematic
posteriors.
* max_treedepth: The maximum depth of the trajectory tree.
* step_scale: float, default 0.25
The initial guess for the step size scaled down by `1/n**(1/4)`.
You can find a full list of arguments in the docstring of the step methods.
Examples
--------
.. code:: ipython
>>> import pymc3 as pm
... n = 100
... h = 61
... alpha = 2
... beta = 2
.. code:: ipython
>>> with pm.Model() as model: # context management
... p = pm.Beta('p', alpha=alpha, beta=beta)
... y = pm.Binomial('y', n=n, p=p, observed=h)
... trace = pm.sample(2000, tune=1000, cores=4)
>>> pm.summary(trace)
mean sd mc_error hpd_2.5 hpd_97.5
p 0.604625 0.047086 0.00078 0.510498 0.694774
"""
model = modelcontext(model)
nuts_kwargs = kwargs.pop('nuts_kwargs', None)
if nuts_kwargs is not None:
warnings.warn("The nuts_kwargs argument has been deprecated. Pass step "
"method arguments directly to sample instead",
DeprecationWarning)
kwargs.update(nuts_kwargs)
step_kwargs = kwargs.pop('step_kwargs', None)
if step_kwargs is not None:
warnings.warn("The step_kwargs argument has been deprecated. Pass step "
"method arguments directly to sample instead",
DeprecationWarning)
kwargs.update(step_kwargs)
if cores is None:
cores = min(4, _cpu_count())
if isinstance(step, pm.step_methods.smc.SMC):
trace = smc.sample_smc(draws=draws,
step=step,
start=start,
cores=cores,
progressbar=progressbar,
model=model,
random_seed=random_seed)
else:
if 'njobs' in kwargs:
cores = kwargs['njobs']
warnings.warn(
"The njobs argument has been deprecated. Use cores instead.",
DeprecationWarning)
if 'nchains' in kwargs:
chains = kwargs['nchains']
warnings.warn(
"The nchains argument has been deprecated. Use chains instead.",
DeprecationWarning)
if chains is None:
chains = max(2, cores)
if isinstance(start, dict):
start = [start] * chains
if random_seed == -1:
random_seed = None
if chains == 1 and isinstance(random_seed, int):
random_seed = [random_seed]
if random_seed is None or isinstance(random_seed, int):
if random_seed is not None:
np.random.seed(random_seed)
random_seed = [np.random.randint(2 ** 30) for _ in range(chains)]
if not isinstance(random_seed, Iterable):
raise TypeError(
'Invalid value for `random_seed`. Must be tuple, list or int')
if 'chain' in kwargs:
chain_idx = kwargs['chain']
warnings.warn(
"The chain argument has been deprecated. Use chain_idx instead.",
DeprecationWarning)
if start is not None:
for start_vals in start:
_check_start_shape(model, start_vals)
# small trace warning
if draws == 0:
msg = "Tuning was enabled throughout the whole trace."
_log.warning(msg)
elif draws < 500:
msg = "Only %s samples in chain." % draws
_log.warning(msg)
draws += tune
if model.ndim == 0:
raise ValueError('The model does not contain any free variables.')
if step is None and init is not None and all_continuous(model.vars):
try:
# By default, try to use NUTS
_log.info('Auto-assigning NUTS sampler...')
start_, step = init_nuts(init=init, chains=chains, n_init=n_init,
model=model, random_seed=random_seed,
progressbar=progressbar, **kwargs)
if start is None:
start = start_
except (AttributeError, NotImplementedError, tg.NullTypeGradError):
# gradient computation failed
_log.info("Initializing NUTS failed. "
"Falling back to elementwise auto-assignment.")
_log.debug('Exception in init nuts', exec_info=True)
step = assign_step_methods(model, step, step_kwargs=kwargs)
else:
step = assign_step_methods(model, step, step_kwargs=kwargs)
if isinstance(step, list):
step = CompoundStep(step)
if start is None:
start = {}
if isinstance(start, dict):
start = [start] * chains
sample_args = {'draws': draws,
'step': step,
'start': start,
'trace': trace,
'chain': chain_idx,
'chains': chains,
'tune': tune,
'progressbar': progressbar,
'model': model,
'random_seed': random_seed,
'cores': cores, }
sample_args.update(kwargs)
has_population_samplers = np.any([isinstance(m, arraystep.PopulationArrayStepShared)
for m in (step.methods if isinstance(step, CompoundStep) else [step])])
parallel = cores > 1 and chains > 1 and not has_population_samplers
if parallel:
_log.info('Multiprocess sampling ({} chains in {} jobs)'.format(chains, cores))
_print_step_hierarchy(step)
try:
trace = _mp_sample(**sample_args)
except pickle.PickleError:
_log.warning("Could not pickle model, sampling singlethreaded.")
_log.debug('Pickling error:', exec_info=True)
parallel = False
except AttributeError as e:
if str(e).startswith("AttributeError: Can't pickle"):
_log.warning("Could not pickle model, sampling singlethreaded.")
_log.debug('Pickling error:', exec_info=True)
parallel = False
else:
raise
if not parallel:
if has_population_samplers:
_log.info('Population sampling ({} chains)'.format(chains))
_print_step_hierarchy(step)
trace = _sample_population(**sample_args, parallelize=cores > 1)
else:
_log.info('Sequential sampling ({} chains in 1 job)'.format(chains))
_print_step_hierarchy(step)
trace = _sample_many(**sample_args)
discard = tune if discard_tuned_samples else 0
trace = trace[discard:]
if compute_convergence_checks:
if draws-tune < 100:
warnings.warn("The number of samples is too small to check convergence reliably.")
else:
trace.report._run_convergence_checks(trace, model)
trace.report._log_summary()
return trace
def _check_start_shape(model, start):
if not isinstance(start, dict):
raise TypeError("start argument must be a dict or an array-like of dicts")
e = ''
for var in model.vars:
if var.name in start.keys():
var_shape = var.shape.tag.test_value
start_var_shape = np.shape(start[var.name])
if start_var_shape:
if not np.array_equal(var_shape, start_var_shape):
e += "\nExpected shape {} for var '{}', got: {}".format(
tuple(var_shape), var.name, start_var_shape
)
# if start var has no shape
else:
# if model var has a specified shape
if var_shape.size > 0:
e += "\nExpected shape {} for var " \
"'{}', got scalar {}".format(
tuple(var_shape), var.name, start[var.name]
)
if e != '':
raise ValueError("Bad shape for start argument:{}".format(e))
def _sample_many(draws, chain, chains, start, random_seed, step, **kwargs):
traces = []
for i in range(chains):
trace = _sample(draws=draws, chain=chain + i, start=start[i],
step=step, random_seed=random_seed[i], **kwargs)
if trace is None:
if len(traces) == 0:
raise ValueError('Sampling stopped before a sample was created.')
else:
break
elif len(trace) < draws:
if len(traces) == 0:
traces.append(trace)
break
else:
traces.append(trace)
return MultiTrace(traces)
def _sample_population(draws, chain, chains, start, random_seed, step, tune,
model, progressbar=None, parallelize=False, **kwargs):
# create the generator that iterates all chains in parallel
chains = [chain + c for c in range(chains)]
sampling = _prepare_iter_population(draws, chains, step, start, parallelize,
tune=tune, model=model, random_seed=random_seed)
if progressbar:
sampling = tqdm(sampling, total=draws)
latest_traces = None
for it, traces in enumerate(sampling):
latest_traces = traces
return MultiTrace(latest_traces)
def _sample(chain, progressbar, random_seed, start, draws=None, step=None,
trace=None, tune=None, model=None, **kwargs):
skip_first = kwargs.get('skip_first', 0)
sampling = _iter_sample(draws, step, start, trace, chain,
tune, model, random_seed)
_pbar_data = None
if progressbar:
_pbar_data = {"chain": chain, "divergences": 0}
_desc = "Sampling chain {chain:d}, {divergences:,d} divergences"
sampling = tqdm(sampling, total=draws, desc=_desc.format(**_pbar_data))
try:
strace = None
for it, (strace, diverging) in enumerate(sampling):
if it >= skip_first:
trace = MultiTrace([strace])
if diverging and _pbar_data is not None:
_pbar_data["divergences"] += 1
sampling.set_description(_desc.format(**_pbar_data))
except KeyboardInterrupt:
pass
finally:
if progressbar:
sampling.close()
return strace
def iter_sample(draws, step, start=None, trace=None, chain=0, tune=None,
model=None, random_seed=None):
"""Generator that returns a trace on each iteration using the given
step method. Multiple step methods supported via compound step
method returns the amount of time taken.
Parameters
----------
draws : int
The number of samples to draw
step : function
Step function
start : dict
Starting point in parameter space (or partial point). Defaults to trace.point(-1)) if
there is a trace provided and model.test_point if not (defaults to empty dict)
trace : backend, list, or MultiTrace
This should be a backend instance, a list of variables to track, or a MultiTrace object
with past values. If a MultiTrace object is given, it must contain samples for the chain
number `chain`. If None or a list of variables, the NDArray backend is used.
chain : int
Chain number used to store sample in backend. If `cores` is greater than one, chain numbers
will start here.
tune : int
Number of iterations to tune, if applicable (defaults to None)
model : Model (optional if in `with` context)
random_seed : int or list of ints
A list is accepted if more if `cores` is greater than one.
Examples
--------
::
for trace in iter_sample(500, step):
...
"""
sampling = _iter_sample(draws, step, start, trace, chain, tune,
model, random_seed)
for i, (strace, _) in enumerate(sampling):
yield MultiTrace([strace[:i + 1]])
def _iter_sample(draws, step, start=None, trace=None, chain=0, tune=None,
model=None, random_seed=None):
model = modelcontext(model)
draws = int(draws)
if random_seed is not None:
np.random.seed(random_seed)
if draws < 1:
raise ValueError('Argument `draws` must be greater than 0.')
if start is None:
start = {}
strace = _choose_backend(trace, chain, model=model)
if len(strace) > 0:
update_start_vals(start, strace.point(-1), model)
else:
update_start_vals(start, model.test_point, model)
try:
step = CompoundStep(step)
except TypeError:
pass
point = Point(start, model=model)
if step.generates_stats and strace.supports_sampler_stats:
strace.setup(draws, chain, step.stats_dtypes)
else:
strace.setup(draws, chain)
try:
step.tune = bool(tune)
for i in range(draws):
if i == tune:
step = stop_tuning(step)
if step.generates_stats:
point, stats = step.step(point)
if strace.supports_sampler_stats:
strace.record(point, stats)
diverging = i > tune and stats and stats[0].get('diverging')
else:
strace.record(point)
else:
point = step.step(point)
strace.record(point)
diverging = False
yield strace, diverging
except KeyboardInterrupt:
strace.close()
if hasattr(step, 'warnings'):
warns = step.warnings()
strace._add_warnings(warns)
raise
except BaseException:
strace.close()
raise
else:
strace.close()
if hasattr(step, 'warnings'):
warns = step.warnings()
strace._add_warnings(warns)
class PopulationStepper:
def __init__(self, steppers, parallelize):
"""Tries to use multiprocessing to parallelize chains.
Falls back to sequential evaluation if multiprocessing fails.
In the multiprocessing mode of operation, a new process is started for each
chain/stepper and Pipes are used to communicate with the main process.
Parameters
----------
steppers : list
A collection of independent step methods, one for each chain.
parallelize : bool
Indicates if chain parallelization is desired
"""
self.nchains = len(steppers)
self.is_parallelized = False
self._master_ends = []
self._processes = []
self._steppers = steppers
if parallelize:
try:
# configure a child process for each stepper
_log.info('Attempting to parallelize chains to all cores. You can turn this off with `pm.sample(cores=1)`.')
import multiprocessing
for c, stepper in enumerate(tqdm(steppers)):
slave_end, master_end = multiprocessing.Pipe()
stepper_dumps = pickle.dumps(stepper, protocol=4)
process = multiprocessing.Process(
target=self.__class__._run_slave,
args=(c, stepper_dumps, slave_end),
name='ChainWalker{}'.format(c)
)
# we want the child process to exit if the parent is terminated
process.daemon = True
# Starting the process might fail and takes time.
# By doing it in the constructor, the sampling progress bar
# will not be confused by the process start.
process.start()
self._master_ends.append(master_end)
self._processes.append(process)
self.is_parallelized = True
except Exception:
_log.info('Population parallelization failed. '
'Falling back to sequential stepping of chains.')
_log.debug('Error was: ', exec_info=True)
else:
_log.info('Chains are not parallelized. You can enable this by passing '
'`pm.sample(cores=n)`, where n > 1.')
return super().__init__()
def __enter__(self):
"""Does nothing because processes are already started in __init__."""
return
def __exit__(self, exc_type, exc_val, exc_tb):
if len(self._processes) > 0:
try:
for master_end in self._master_ends:
master_end.send(None)
for process in self._processes:
process.join(timeout=3)
except Exception:
_log.warning('Termination failed.')
return
@staticmethod
def _run_slave(c, stepper_dumps, slave_end):
"""Started on a separate process to perform stepping of a chain.
Parameters
----------
c : int
number of this chain
stepper : BlockedStep
a step method such as CompoundStep
slave_end : multiprocessing.connection.PipeConnection
This is our connection to the main process
"""
# re-seed each child process to make them unique
np.random.seed(None)
try:
stepper = pickle.loads(stepper_dumps)
# the stepper is not necessarily a PopulationArraySharedStep itself,
# but rather a CompoundStep. PopulationArrayStepShared.population
# has to be updated, therefore we identify the substeppers first.
population_steppers = []
for sm in (stepper.methods if isinstance(stepper, CompoundStep) else [stepper]):
if isinstance(sm, arraystep.PopulationArrayStepShared):
population_steppers.append(sm)
while True:
incoming = slave_end.recv()
# receiving a None is the signal to exit
if incoming is None:
break
tune_stop, population = incoming
if tune_stop:
stop_tuning(stepper)
# forward the population to the PopulationArrayStepShared objects
# This is necessary because due to the process fork, the population
# object is no longer shared between the steppers.
for popstep in population_steppers:
popstep.population = population
update = stepper.step(population[c])
slave_end.send(update)
except Exception:
_log.exception('ChainWalker{}'.format(c))
return
def step(self, tune_stop, population):
"""Steps the entire population of chains.
Parameters
----------
tune_stop : bool
Indicates if the condition (i == tune) is fulfilled
population : list
Current Points of all chains
Returns
-------
update : Point
The new positions of the chains
"""
updates = [None] * self.nchains
if self.is_parallelized:
for c in range(self.nchains):
self._master_ends[c].send((tune_stop, population))
# Blockingly get the step outcomes
for c in range(self.nchains):
updates[c] = self._master_ends[c].recv()
else:
for c in range(self.nchains):
if tune_stop:
self._steppers[c] = stop_tuning(self._steppers[c])
updates[c] = self._steppers[c].step(population[c])
return updates
def _prepare_iter_population(draws, chains, step, start, parallelize, tune=None,
model=None, random_seed=None):
"""Prepares a PopulationStepper and traces for population sampling.
Returns
-------
_iter_population : generator
The generator the yields traces of all chains at the same time
"""
# chains contains the chain numbers, but for indexing we need indices...
nchains = len(chains)
model = modelcontext(model)
draws = int(draws)
if random_seed is not None:
np.random.seed(random_seed)
if draws < 1:
raise ValueError('Argument `draws` should be above 0.')
# The initialization of traces, samplers and points must happen in the right order:
# 1. traces are initialized and update_start_vals configures variable transforms
# 2. population of points is created
# 3. steppers are initialized and linked to the points object
# 4. traces are configured to track the sampler stats
# 5. a PopulationStepper is configured for parallelized stepping
# 1. prepare a BaseTrace for each chain
traces = [_choose_backend(None, chain, model=model) for chain in chains]
for c, strace in enumerate(traces):
# initialize the trace size and variable transforms
if len(strace) > 0:
update_start_vals(start[c], strace.point(-1), model)
else:
update_start_vals(start[c], model.test_point, model)
# 2. create a population (points) that tracks each chain
# it is updated as the chains are advanced
population = [Point(start[c], model=model) for c in range(nchains)]
# 3. Set up the steppers
steppers = [None] * nchains
for c in range(nchains):
# need indepenent samplers for each chain
# it is important to copy the actual steppers (but not the delta_logp)
if isinstance(step, CompoundStep):
chainstep = CompoundStep([copy(m) for m in step.methods])
else:
chainstep = copy(step)
# link population samplers to the shared population state
for sm in (chainstep.methods if isinstance(step, CompoundStep) else [chainstep]):
if isinstance(sm, arraystep.PopulationArrayStepShared):
sm.link_population(population, c)
steppers[c] = chainstep
# 4. configure tracking of sampler stats
for c in range(nchains):
if steppers[c].generates_stats and traces[c].supports_sampler_stats:
traces[c].setup(draws, c, steppers[c].stats_dtypes)
else:
traces[c].setup(draws, c)
# 5. configure the PopulationStepper (expensive call)
popstep = PopulationStepper(steppers, parallelize)
# Because the preparations above are expensive, the actual iterator is
# in another method. This way the progbar will not be disturbed.
return _iter_population(draws, tune, popstep, steppers, traces, population)
def _iter_population(draws, tune, popstep, steppers, traces, points):
"""Generator that iterates a PopulationStepper.
Parameters
----------
draws : int
number of draws per chain
tune : int
number of tuning steps
popstep : PopulationStepper
the helper object for (parallelized) stepping of chains
steppers : list
The step methods for each chain
traces : list
Traces for each chain
points : list
population of chain states
"""
try:
with popstep:
# iterate draws of all chains
for i in range(draws):
updates = popstep.step(i == tune, points)
# apply the update to the points and record to the traces
for c, strace in enumerate(traces):
if steppers[c].generates_stats:
points[c], stats = updates[c]
if strace.supports_sampler_stats:
strace.record(points[c], stats)
else:
strace.record(points[c])
else:
points[c] = updates[c]
strace.record(points[c])
# yield the state of all chains in parallel
yield traces
except KeyboardInterrupt:
for c, strace in enumerate(traces):
strace.close()
if hasattr(steppers[c], 'report'):
steppers[c].report._finalize(strace)
raise
except BaseException:
for c, strace in enumerate(traces):
strace.close()
raise
else:
for c, strace in enumerate(traces):
strace.close()
if hasattr(steppers[c], 'report'):
steppers[c].report._finalize(strace)
def _choose_backend(trace, chain, shortcuts=None, **kwds):
if isinstance(trace, BaseTrace):
return trace
if isinstance(trace, MultiTrace):
return trace._straces[chain]
if trace is None:
return NDArray(**kwds)
if shortcuts is None:
shortcuts = pm.backends._shortcuts
try:
backend = shortcuts[trace]['backend']
name = shortcuts[trace]['name']
return backend(name, **kwds)
except TypeError:
return NDArray(vars=trace, **kwds)
except KeyError:
raise ValueError('Argument `trace` is invalid.')
def _mp_sample(draws, tune, step, chains, cores, chain, random_seed,
start, progressbar, trace=None, model=None, **kwargs):
import pymc3.parallel_sampling as ps
# We did draws += tune in pm.sample
draws -= tune
traces = []
for idx in range(chain, chain + chains):
if trace is not None:
strace = _choose_backend(copy(trace), idx, model=model)
else:
strace = _choose_backend(None, idx, model=model)
# for user supply start value, fill-in missing value if the supplied
# dict does not contain all parameters
update_start_vals(start[idx - chain], model.test_point, model)
if step.generates_stats and strace.supports_sampler_stats:
strace.setup(draws + tune, idx + chain, step.stats_dtypes)
else:
strace.setup(draws + tune, idx + chain)
traces.append(strace)
sampler = ps.ParallelSampler(
draws, tune, chains, cores, random_seed, start, step,
chain, progressbar)
try:
try:
with sampler:
for draw in sampler:
trace = traces[draw.chain - chain]
if (trace.supports_sampler_stats
and draw.stats is not None):
trace.record(draw.point, draw.stats)
else:
trace.record(draw.point)
if draw.is_last:
trace.close()
if draw.warnings is not None:
trace._add_warnings(draw.warnings)
except ps.ParallelSamplingError as error:
trace = traces[error._chain - chain]
trace._add_warnings(error._warnings)
for trace in traces:
trace.close()
multitrace = MultiTrace(traces)
multitrace._report._log_summary()
raise
return MultiTrace(traces)
except KeyboardInterrupt: