Skip to content

API Reference: Models

Bases: ABC

Abstract class representing an economic model.

Source code in src/dyno/model.py
 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
class AbstractModel(ABC):
    """Abstract class representing an economic model."""

    name: str | None
    filename: str
    context: ModelContext
    symbols: dict[str, list[str]]
    processes: ProductNormal | None
    paths: dict[str, dict[int, float]] | None
    symbolic: SymbolicModel
    __steady_state__: dict[str, float] | None
    strict: bool
    _invalid_shifts: list[str] | None
    _steady_stats: dict[str, Any] | None = None
    _calibration_overrides: dict[str, Any]

    def __init__(
        self: Self,
        filename: str | os.PathLike[str] | None = None,
        txt: str | None = None,
        strict: bool = False,
        **kwargs: Any,
    ) -> None:
        self.strict = strict
        self._invalid_shifts = None
        if filename is not None:
            filename = os.fspath(filename)

        match filename, txt:
            case (None, None):
                raise ValueError(
                    "Neither the file name nor content were passed to constructor. One of the two should be passed."
                )
            case (None, txt):
                assert txt is not None
                self.filename = "*anonymous*.dyno"
                self.import_model(txt, **kwargs)
            case (filename, None):
                assert filename is not None
                self.filename = filename
                self.import_file(filename, **kwargs)
            case _:
                assert filename is not None
                assert txt is not None
                self.filename = filename
                self.import_model(txt, **kwargs)

        self.__steady_state__ = None
        self._set_name()
        self._set_context()
        self._set_symbols()
        self._set_exogenous()

    def copy(self: Self):
        import copy

        return copy.deepcopy(self)

    def recalibrate(self: Self, **calib: Any) -> Self:
        """Return a new model instance with updated parameter/steady-state calibration."""
        raise NotImplementedError

    def variants(self: Self, *args: Any, **kwargs: Any) -> "VariantCollection[Self]":
        """Create a ``VariantCollection`` holding differently calibrated versions of this model.

        Examples
        --------
        >>> var = model.variants(a=[2, 3, 4])
        >>> sims = model.variants(a=[2, 3, 4]).solve().simulate()
        >>> fig = sims.plot()
        """
        from .variants import VariantCollection, _expand_variant_specs

        specs, labels = _expand_variant_specs(args, kwargs)
        models = [self.recalibrate(**spec) for spec in specs]
        return VariantCollection(models, specs=specs, labels=labels)

    def _variables_used_in_equations(self: Self) -> set[str]:
        names: set[str] = set()
        symbolic = getattr(self, "symbolic", None)
        eqs = getattr(symbolic, "equations", []) if symbolic is not None else []
        for eq in eqs:
            if hasattr(eq, "iter_subtrees_topdown"):
                for subtree in eq.iter_subtrees_topdown():
                    if getattr(subtree, "data", None) == "variable":
                        if subtree.children and hasattr(
                            subtree.children[0], "children"
                        ):
                            names.add(str(subtree.children[0].children[0]))
        return names

    def _repr_data(self: Self) -> dict[str, Any]:
        return model_repr_data(self)

    def _render_repr_text(self: Self, data: dict[str, Any]) -> str:
        return render_model_text(data)

    def _render_repr_html(self: Self, data: dict[str, Any]) -> str:
        return render_model_html(data)

    def __repr__(self: Self) -> str:
        return self._render_repr_text(self._repr_data())

    @property
    def checks(self) -> dict[str, bool]:
        return {"deterministic": self.processes is None}

    @property
    def metadata(self) -> dict[str, Any]:
        """Convenience accessor for model metadata parsed from source files."""
        symbolic = getattr(self, "symbolic", None)
        if symbolic is None:
            return {}
        return getattr(symbolic, "metadata", {})

    @metadata.setter
    def metadata(self, value: dict[str, Any]) -> None:
        symbolic = getattr(self, "symbolic", None)
        if symbolic is None:
            raise AttributeError(
                "Cannot set metadata before symbolic model is initialized"
            )
        symbolic.metadata = value
        if (
            isinstance(value, dict)
            and "name" in value
            and isinstance(value["name"], str)
        ):
            self.name = value["name"]

    @property
    def is_deterministic(self) -> bool:
        return self.processes is None

    def equations_with_tags(self: Self) -> list[tuple[int, str, list[str]]]:
        """Return equations paired with metadata tags and key/value annotations."""

        symbolic = getattr(self, "symbolic", None)
        if symbolic is None:
            return []

        pairs: list[tuple[Any, Any]]
        iterator = getattr(symbolic, "iter_equations_with_metadata", None)
        if callable(iterator):
            pairs = list(iterator())
        else:
            equations = list(getattr(symbolic, "equations", []))
            pairs = []
            for eq in equations:
                meta = getattr(eq, "meta", None)
                statement_metadata = (
                    getattr(meta, "statement_metadata", {}) if meta is not None else {}
                )
                pairs.append((eq, statement_metadata))

        rows: list[tuple[int, str, list[str]]] = []
        for i, (eq, meta) in enumerate(pairs, start=1):
            eq_text = str(eq)
            try:
                from dyno.dynspec.grammar import str_expression

                eq_text = str_expression(eq)
            except Exception:
                pass

            tags: list[str] = []
            if isinstance(meta, dict):
                raw_tags = meta.get("tags", [])
                if isinstance(raw_tags, list):
                    tags = [str(t) for t in raw_tags]
                for k, v in meta.items():
                    if k != "tags":
                        tags.append(f"{k}={v}")

            rows.append((i, eq_text, tags))

        return rows

    def print_equations_with_tags(self: Self) -> None:
        """Print all equations with their tags."""

        for i, eq_text, tags in self.equations_with_tags():
            tags_txt = ", ".join(tags) if tags else "-"
            print(f"{i:>3}. {eq_text}  [tags: {tags_txt}]")

    @abstractmethod
    def import_model(self: Self, txt: str, **kwargs: Any) -> None:
        """Set `symbolic` attribute from model text description."""

    @abstractmethod
    def _set_context(self: Self) -> None: ...

    @abstractmethod
    def compute_residuals(self: Self, y2, y1, y0, e) -> TVector: ...

    @abstractmethod
    def compute_jacobians(
        self: Self, y2, y1, y0, e
    ) -> tuple[TVector, TMatrix, TMatrix, TMatrix, TMatrix]: ...

    def run(self: Self, default_pipeline: bool = False) -> "RunResults":
        from .report import RunResults

        return RunResults(model=self)

    def import_file(
        self: Self, filename: str | os.PathLike[str], **kwargs: Any
    ) -> None:
        with open(filename, "rt", encoding="utf-8") as f:
            txt = f.read()
        self.import_model(txt, **kwargs)

    def _set_name(self: Self) -> None:
        import os.path

        symbolic = getattr(self, "symbolic", None)
        meta = getattr(symbolic, "metadata", {}) if symbolic is not None else {}
        name = meta.get("name") if isinstance(meta, dict) else None
        if name and isinstance(name, str):
            self.name = name
            return

        if self.filename:
            self.name = os.path.basename(self.filename).split(".")[0]
        else:
            self.name = "Unnamed"

    def _set_symbols(self: Self) -> None:
        c = self.context

        # exogenous are either defined as processes or by specifying values with t >= 1
        # Preserve declaration/insertion order (avoid set() which scrambles order)
        exo_order: list[str] = []

        # Variables declared exogenous by the source file (``varexo`` in .mod
        # files) are exogenous even without a process or a forced path.
        declared = getattr(self.symbolic, "symbols", None)
        if isinstance(declared, dict):
            for name in declared.get("exogenous", []):
                if name not in exo_order:
                    exo_order.append(name)

        for exo_tuple in c.get("processes", {}).keys():
            for name in exo_tuple:
                if name not in exo_order:
                    exo_order.append(name)

        for name, val in c.get("values", {}).items():
            if isinstance(val, dict):
                has_t_ge_1 = any(
                    isinstance(t, (int, float)) and t >= 1 for t in val.keys()
                )
            elif isinstance(val, (list, tuple, np.ndarray)):
                has_t_ge_1 = len(val) > 0
            else:
                has_t_ge_1 = False

            if has_t_ge_1 and name not in exo_order:
                exo_order.append(name)

        exo_set = set(exo_order)

        declared_vars = list(c.get("variables", {}).keys())
        variables: list[str] = []
        for v in declared_vars:
            if v not in exo_set:
                variables.append(v)
        for name in c.get("values", {}).keys():
            if name not in exo_set and name not in variables:
                variables.append(name)
        for v in declared_vars:
            if v in exo_set and v not in variables:
                variables.append(v)
        for v in exo_order:
            if v not in variables:
                variables.append(v)

        exogenous = [v for v in variables if v in exo_set]
        endogenous = [v for v in variables if v not in exogenous]

        # try:
        #     parameters = [*self.symbolic.parameters]
        # except Exception:
        parameters = [*c["constants"].keys()]

        self.symbols = {
            "variables": variables,
            "endogenous": endogenous,
            "exogenous": exogenous,
            "parameters": parameters,
        }

    def _set_exogenous(self: Self) -> None:
        pps = self.context["processes"].values()
        if len(pps) == 0:
            self.processes = None
        else:
            self.processes = ProductNormal(*[e for e in pps])

    @property
    def steady_state(self) -> dict[str, float]:
        if self.__steady_state__ is None:
            self.__steady_state__ = self.context["steady_states"]
        return self.__steady_state__

    @property
    def residuals(self):
        y, e = self.__steady_state_vectors__
        return self.compute_residuals(y, y, y, e)

    def _check_defined(self: Self, action: str) -> None:
        """Raise ``UndefinedSymbolError`` if a parameter or a steady state is still NaN."""
        constants = self.context.get("constants", {})
        steady_states = self.context.get("steady_states", {})
        unassigned_params = [
            p
            for p in self.symbols.get("parameters", [])
            if isinstance(val := constants.get(p), float) and np.isnan(val)
        ]
        unassigned_ss = [
            v
            for v in self.symbols.get("endogenous", [])
            if isinstance(val := steady_states.get(v), float) and np.isnan(val)
        ]
        if unassigned_params or unassigned_ss:
            msgs = []
            if unassigned_params:
                msgs.append(
                    f"parameters without values: {', '.join(sorted(unassigned_params))}"
                )
            if unassigned_ss:
                msgs.append(
                    f"variables without steady state: {', '.join(sorted(unassigned_ss))}"
                    " (declare them with `x[~] <- ...` or call `model.steady()`)"
                )
            from .errors import UndefinedSymbolError

            raise UndefinedSymbolError(
                f"Cannot {action} due to uninitialized symbols ({'; '.join(msgs)})."
            )

    _check_eigenvalues: bool = False

    def check(
        self: "Self", tol: float = 1e-6, compute_eigenvalues: bool | None = None
    ) -> "Self":
        self._check_defined("check model")

        r = self.residuals
        if not all(abs(x) < tol for x in r):
            raise SteadyStateError(r)
        if compute_eigenvalues is None:
            compute_eigenvalues = self._check_eigenvalues
        if compute_eigenvalues:
            from .errors import BlanchardKahnError
            from .solver import solve_qz

            jac = self.jacobians
            A, B, C = jac[1], jac[2], jac[3]
            if A.shape[0] == 0:
                self._eigenvalues = None
            else:
                try:
                    _, evs = solve_qz(A, B, C)
                except BlanchardKahnError as e:
                    # Keep the eigenvalues: they show why the conditions fail.
                    evs = e.evs
                except Exception:
                    evs = None
                self._eigenvalues = evs
        else:
            self._eigenvalues = None
        return self

    @property
    def steady_stats(self: Self) -> dict[str, Any] | None:
        """Convergence statistics and algorithm details from the most recent steady-state calculation."""
        return getattr(self, "_steady_stats", None)

    def _variable_occurrences(self: Self) -> dict[str, tuple[int, int, int | None]]:
        """Map each variable to ``(n_equations, n_occurrences, first_line)``.

        ``n_equations`` counts the equations a variable appears in,
        ``n_occurrences`` counts every ``name[...]`` reference, and
        ``first_line`` is the source line of the first reference (if known).
        Only parse trees (``DynoFile``/``LModFile`` equations) are inspected;
        other equation representations yield an empty mapping.
        """
        occurrences: dict[str, tuple[int, int, int | None]] = {}
        for eq in getattr(self.symbolic, "equations", []):
            iter_subtrees = getattr(eq, "iter_subtrees_topdown", None)
            if not callable(iter_subtrees):
                continue
            eq_line = getattr(getattr(eq, "meta", None), "line", None)
            in_eq: set[str] = set()
            for subtree in iter_subtrees():
                if getattr(subtree, "data", None) not in ("variable", "value"):
                    continue
                try:
                    name = str(subtree.children[0].children[0])
                except (AttributeError, IndexError):
                    continue
                n_eq, n_occ, line = occurrences.get(name, (0, 0, None))
                if name not in in_eq:
                    in_eq.add(name)
                    n_eq += 1
                if line is None:
                    line = getattr(getattr(subtree, "meta", None), "line", None)
                    line = line if line is not None else eq_line
                occurrences[name] = (n_eq, n_occ + 1, line)
        return occurrences

    def _suspicious_endogenous(
        self: Self, require_single_occurrence: bool = False
    ) -> list[str]:
        """Describe endogenous variables that look like typos.

        By default a variable is reported when it has no steady-state value
        or appears in at most one equation; if some variables lack a steady
        state, only those are reported. With ``require_single_occurrence``,
        only variables referenced exactly once in the whole model *and*
        lacking a steady state are reported. Each description suggests the
        closest declared symbol (variables with a steady state, parameters,
        exogenous).
        """
        import difflib
        import math

        endogenous = list(self.symbols.get("endogenous", []))
        steady_states = self.context.get("steady_states", {})
        # Empty when equations are not parse trees (e.g. DynareModel): then
        # occurrence counts are unknown and only steady states are used.
        occurrences = self._variable_occurrences()

        def has_steady_state(name: str) -> bool:
            value = steady_states.get(name)
            if value is None:
                return False
            try:
                return not math.isnan(float(value))
            except (TypeError, ValueError):
                return True

        declared = [v for v in endogenous if has_steady_state(v)]
        declared += list(self.symbols.get("parameters", []))
        declared += list(self.symbols.get("exogenous", []))

        flagged: list[tuple[bool, str]] = []
        for name in endogenous:
            missing_ss = not has_steady_state(name)
            if occurrences:
                n_eq, n_occ, line = occurrences.get(name, (0, 0, None))
            else:
                n_eq, n_occ, line = 2, 2, None  # unknown: never "rare"
            if require_single_occurrence:
                if not (missing_ss and n_occ == 1):
                    continue
            elif not (missing_ss or n_eq <= 1):
                continue

            line_txt = f" (line {line})" if line is not None else ""
            facts: list[str] = []
            if n_occ == 0:
                facts.append("does not appear in any equation")
            elif n_occ == 1:
                facts.append(f"appears only once{line_txt}")
            elif n_eq == 1:
                facts.append(f"appears in a single equation{line_txt}")
            if missing_ss:
                facts.append("has no steady state")
            desc = f"'{name}' " + " and ".join(facts) + "."
            candidates = [c for c in dict.fromkeys(declared) if c != name]
            matches = difflib.get_close_matches(name, candidates, n=1)
            if matches:
                desc += f" Did you mean '{matches[0]}'?"
            flagged.append((missing_ss, desc))

        # Variables without steady state are the most likely culprits: when
        # there are any, the single-equation ones are only noise.
        if any(missing_ss for missing_ss, _ in flagged):
            flagged = [item for item in flagged if item[0]]
        return [desc for _, desc in flagged]

    def _check_square(self: Self) -> None:
        """Raise ``SystemStructureError`` unless there are as many equations as endogenous variables."""
        neq = len(getattr(self.symbolic, "equations", []))
        endogenous = list(self.symbols.get("endogenous", []))
        n_endo = len(endogenous)
        if neq != n_endo:
            from .errors import SystemStructureError

            def plural(n: int, word: str) -> str:
                return f"{n} {word}" if n == 1 else f"{n} {word}s"

            lines = [
                f"Model has {plural(neq, 'equation')} but "
                f"{plural(n_endo, 'endogenous variable')}. "
                "The dynamic system must be square."
            ]
            hints = self._suspicious_endogenous()
            max_hints = 10
            lines += [f"  {h}" for h in hints[:max_hints]]
            if len(hints) > max_hints:
                lines.append(f"  ... and {len(hints) - max_hints} more.")
            if not hints:
                lines.append(f"  Endogenous variables: {', '.join(endogenous)}")
            raise SystemStructureError("\n".join(lines))

    def steady(
        self: Self,
        tol: float = 1e-10,
        maxiter: int = 100,
        method: str = "hybr",
        **options: Any,
    ) -> Self:
        from scipy.optimize import root

        invalid_shifts = getattr(self, "_invalid_shifts", None)
        if invalid_shifts:
            from .errors import SystemStructureError

            shifts_str = ", ".join(invalid_shifts)
            raise SystemStructureError(
                f"Unsupported timing (only -1, 0, 1 so far): higher-order lead/lag detected ({shifts_str}). "
                "Dyno solvers currently support shifts in [-1, 1]."
            )

        if "tol" in options:
            tol = float(options.pop("tol"))
        if "maxiter" in options:
            maxiter = int(options.pop("maxiter"))
        if "method" in options:
            method = str(options.pop("method"))

        self._check_square()

        endogenous = self.symbols["endogenous"]
        if len(endogenous) == 0:
            res = self.copy()
            res._steady_stats = {
                "algorithm": method,
                "converged": True,
                "iterations": 0,
                "function_evaluations": 0,
                "jacobian_evaluations": 0,
                "max_residual": 0.0,
                "tolerance": float(tol),
                "message": "No endogenous variables to solve.",
            }
            return res

        y0, _ = self.__steady_state_vectors__
        guess = np.nan_to_num(np.asarray(y0, dtype=float), nan=1.0)

        def _candidate(values: np.ndarray) -> AbstractModel:
            calib = {name: float(values[i]) for i, name in enumerate(endogenous)}
            model = self.recalibrate(**calib)
            for name, value in calib.items():
                model.context["steady_states"][name] = value
            model.__steady_state__ = model.context["steady_states"]
            return model

        def _fun(values: np.ndarray) -> np.ndarray:
            model = _candidate(values)
            return np.asarray(model.residuals, dtype=float)

        def _jac(values: np.ndarray) -> np.ndarray:
            model = _candidate(values)
            jac = model.jacobians
            A, B, C = jac[1], jac[2], jac[3]
            return A + B + C

        solver_options: dict[str, Any] = {"maxfev": maxiter}
        if "options" in options and isinstance(options["options"], dict):
            solver_options.update(options.pop("options"))
        solver_options.update(options)

        sol: Any = root(
            _fun,
            guess,
            jac=_jac,
            method=cast(Any, method),
            options=cast(Any, solver_options),
        )

        solved = cast(Self, _candidate(np.asarray(sol.x, dtype=float)))
        residuals = np.asarray(solved.residuals, dtype=float)

        converged = bool(
            sol.success
            and np.isfinite(residuals).all()
            and (np.max(np.abs(residuals)) <= tol)
        )
        iterations = getattr(sol, "nit", None)
        if iterations is None:
            iterations = getattr(sol, "nfev", None)

        stats = {
            "algorithm": method,
            "converged": converged,
            "iterations": iterations,
            "function_evaluations": getattr(sol, "nfev", None),
            "jacobian_evaluations": getattr(sol, "njev", None),
            "max_residual": (
                float(np.max(np.abs(residuals)))
                if np.isfinite(residuals).all()
                else float("nan")
            ),
            "tolerance": float(tol),
            "message": str(getattr(sol, "message", "")),
        }
        solved._steady_stats = stats

        if not converged:
            raise SteadyStateError(residuals, steady_stats=stats)

        return solved

    @property
    def jacobians(self):
        y, e = self.__steady_state_vectors__
        return self.compute_jacobians(y, y, y, e)

    @property
    def __steady_state_vectors__(self) -> tuple[list[float], list[float]]:
        c = self.context
        y = [c["steady_states"].get(name, nan) for name in self.symbols["endogenous"]]

        e: list[float] = []
        for name in self.symbols["exogenous"]:
            if name in c["steady_states"]:
                e.append(c["steady_states"][name])
            elif name in c["values"]:
                e.append(c["values"][name].get(0, 0.0))
            else:
                e.append(nan)
        return y, e

    def describe(self: Self) -> None:
        from rich.console import Console
        from rich.markdown import Markdown

        console = Console()
        console.print(Markdown(self._markdown_()))

    def _markdown_(self: Self) -> str:
        return render_model_markdown(self._repr_data(), self.filename)

    def _repr_html_(self: Self) -> str:
        return self._render_repr_html(self._repr_data())

    def solve(self: Self, **args: Any) -> "PerturbationSolution":
        """Solve the model for its first-order perturbation decision rule.

        Only stochastic models with an invariant state-space decision rule can be solved
        using this method. For deterministic perfect-foresight models, use ``model.simulate()``
        to compute transition trajectories.

        Returns
        -------
        PerturbationSolution
            First-order perturbation solution wrapping the recursive decision rule.

        Raises
        ------
        SystemStructureError
            If the model is deterministic, or if the model equations are not square or
            have unsupported lag/lead timing.
        """
        invalid_shifts = getattr(self, "_invalid_shifts", None)
        if invalid_shifts:
            from .errors import SystemStructureError

            shifts_str = ", ".join(invalid_shifts)
            raise SystemStructureError(
                f"Unsupported timing (only -1, 0, 1 so far): higher-order lead/lag detected ({shifts_str}). "
                "Dyno solvers currently support shifts in [-1, 1]."
            )

        self._check_square()
        if self.is_deterministic:
            from .errors import SystemStructureError

            raise SystemStructureError(
                "Deterministic models do not have a perturbation decision rule and cannot be solved with `model.solve()`. "
                "Use `model.simulate()` instead to compute the perfect-foresight transition trajectory."
            )
        sol = self.perturb(**args)
        self._solution = sol
        return sol

    def simulate(
        self: Self,
        T: int | None = None,
        mode: SimulateMode = "auto",
        N: int = 1,
        units: UnitsType | None = None,
        solve: bool | PerturbationSolution | None = None,
        **args: Any,
    ) -> "SimulationResult":
        """Simulate the model.

        For deterministic models (or when ``mode`` is ``'transition'`` or ``'deterministic'``),
        runs the stacked-time perfect foresight solver and returns a ``TransitionSimulation``.
        For stochastic models, solves the first-order perturbation (or reuses a supplied/cached
        solution) and delegates to ``solution.simulate(...)``.

        Parameters
        ----------
        T : int, optional
            Simulation horizon. Default is 40 for stochastic, or model-configured/40 for deterministic.
        mode : SimulateMode, default 'auto'
            Simulation mode ('auto', 'random', 'irf', 'transition', 'deterministic').
        N : int, default 1
            Number of Monte Carlo simulations (for stochastic random simulation).
        units : UnitsType, optional
            Units for output ('level', 'deviation', 'percent', 'log-deviation').
        solve : bool | PerturbationSolution | None, optional
            For stochastic models, controls how the model is solved:
            - ``None``: reuses an already cached solution on the model if present; otherwise solves automatically.
            - ``True``: forces computing a fresh perturbation solution.
            - ``PerturbationSolution``: uses the pre-computed solution directly without re-solving.
            - ``False``: requires that a solution has already been cached on the model.
        **args : Any
            Additional options passed to the solver or simulation.
        """
        self._check_square()
        if self.is_deterministic or mode in ("transition", "deterministic"):
            from .solver import deterministic_solve

            target_units: UnitsType = "level" if units is None else units
            return deterministic_solve(self, T=T, units=target_units, **args)

        from .solver import PerturbationSolution

        sol: PerturbationSolution | None = None
        if isinstance(solve, PerturbationSolution):
            sol = solve
        elif solve is False:
            sol = getattr(self, "_solution", None)
            if sol is None:
                raise ValueError(
                    "solve=False was passed, but the model has not been solved yet. "
                    "Call model.solve() first or pass solve=True."
                )
        elif solve is True:
            solve_kwargs = {k: v for k, v in args.items() if k in {"method"}}
            sol = self.perturb(**solve_kwargs)
            self._solution = sol
        else:
            # solve is None
            if getattr(self, "_solution", None) is not None:
                sol = self._solution
            else:
                solve_kwargs = {k: v for k, v in args.items() if k in {"method"}}
                sol = self.perturb(**solve_kwargs)
                self._solution = sol

        sim_kwargs = {k: v for k, v in args.items() if k not in {"method"}}
        horizon = 40 if T is None else int(T)
        target_units = "deviation" if units is None else units
        return sol.simulate(T=horizon, mode=mode, N=N, units=target_units, **sim_kwargs)

    def perturb(self: Self, method: Solver = "qz") -> "PerturbationSolution":
        from .solver import PerturbationSolution, RecursiveDecisionRule
        from .solver import solve as solve_quadratic_matrix

        self._check_defined("solve the model")

        r, A, B, C, D = self.jacobians

        X, evs = solve_quadratic_matrix(A, B, C, method=method)

        Y = linsolve(A @ X + B, -D)

        v = self.symbols["endogenous"]
        e = self.symbols["exogenous"]

        assert self.processes is not None
        Σ = self.processes.Σ

        y, _ = self.__steady_state_vectors__
        y0 = np.reshape(y, len(y))

        dr = RecursiveDecisionRule(
            X,
            Y,
            Σ,
            {"endogenous": v, "exogenous": e},
            x0=y0,
            model=self,
        )
        return PerturbationSolution(dr, evs=evs)

    def _dynamic_point(
        self: Self,
        v_prev: TVector,
        v_curr: TVector,
        v_next: TVector,
        diff: bool = False,
    ) -> TVector | tuple[TVector, TTensor]:
        """Evaluate the dynamic equations at one date.

        ``v_prev``, ``v_curr`` and ``v_next`` hold all variables (endogenous
        first, then exogenous, in the order of ``symbols["variables"]``) at
        dates t-1, t and t+1. Returns the residuals, and when ``diff`` is
        true also the Jacobian ``J`` of shape ``(n_equations, n_variables, 3)``
        where the last axis indexes the date (t-1, t, t+1).

        The default implementation relies on ``compute_residuals`` and
        ``compute_jacobians``, which only see the date-t exogenous values.
        """
        q = len(self.symbols["endogenous"])
        y_prev, y_curr, y_next = v_prev[:q], v_curr[:q], v_next[:q]
        e = v_curr[q:]
        if not diff:
            return np.asarray(self.compute_residuals(y_next, y_curr, y_prev, e))
        r, A, B, C, D = self.compute_jacobians(y_next, y_curr, y_prev, e)
        p = len(self.symbols["variables"])
        J = np.zeros((len(r), p, 3))
        J[:, :q, 0] = C
        J[:, :q, 1] = B
        J[:, :q, 2] = A
        J[:, q:, 1] = D
        return np.asarray(r), J

    def _forced_path(self: Self, T: int) -> TMatrix:
        """Steady state at every date, overridden by ``context["values"]``."""
        y, e = self.__steady_state_vectors__
        ss = np.concatenate([y, e])
        v1 = ss[None, :].repeat(T + 1, axis=0)
        for key, value in self.context.get("values", {}).items():
            i = self.symbols["variables"].index(key)
            for a, b in value.items():
                if 0 <= a <= T:
                    v1[a, i] = b
        return v1

    def deterministic_residuals_with_jacobian(
        self: Self,
        v: np.ndarray,
        sparsify: bool = False,
        continuation: str = "stationary",
        growth_rate: Any = None,
        growth_type: str = "geometric",
        **kwargs: Any,
    ) -> tuple[np.ndarray, Any]:
        """Stacked-time residuals and Jacobian of the perfect-foresight system.

        Generic implementation built on ``_dynamic_point``, one date at a
        time. Date 0 is pinned to the forced path (initial condition), the
        trailing exogenous variables follow ``context["values"]`` at every
        date, and the value after the horizon comes from the terminal
        ``continuation`` (see ``_compute_terminal_continuation``).
        """
        from .dyno_model import (
            _build_dense_jacobian,
            _build_sparse_jacobian,
            _compute_terminal_continuation,
        )

        continuation = kwargs.get("terminal_condition", continuation)
        if continuation in ("static", "steady_static"):
            raise NotImplementedError(
                f"Terminal condition {continuation!r} is only available for DynoModel."
            )

        flat = v.ndim == 1
        variables = list(self.symbols["variables"])
        p = len(variables)
        q = len(self.symbols["endogenous"])
        T = int(np.prod(v.shape) / p - 1)
        v = v.reshape((T + 1, p))

        y, e = self.__steady_state_vectors__
        ss = np.concatenate([y, e])
        v_prev_T = v[-2, :] if T >= 1 else v[-1, :]
        v_next_T, alpha_prev, alpha_curr = _compute_terminal_continuation(
            continuation,
            v_prev_T,
            v[-1, :],
            ss,
            variables,
            growth_rate=growth_rate,
            growth_type=growth_type,
        )
        v_b = np.concatenate([v[0, :][None, :], v[:-1, :]], axis=0)
        v_f = np.concatenate([v[1:, :], v_next_T[None, :]], axis=0)

        v1 = self._forced_path(T)
        N = T + 1
        res = np.zeros((N, p))
        DD = np.zeros((N, p, p, 3))
        for t in range(N):
            r_t, J_t = cast(
                tuple[TVector, TTensor],
                self._dynamic_point(v_b[t], v[t], v_f[t], diff=True),
            )
            res[t, :q] = r_t
            DD[t, :q, :, :] = J_t
        res[:, q:] = v[:, q:] - v1[:, q:]
        for i in range(q, p):
            DD[:, i, i, 1] = 1.0
        res[0, :] = v[0, :] - v1[0, :]  # initial condition

        if not flat:
            return res, DD
        J: Any
        if sparsify:
            J = _build_sparse_jacobian(N, p, DD, {}, alpha_prev, alpha_curr)
        else:
            J = _build_dense_jacobian(N, p, DD, {}, alpha_prev, alpha_curr)
        return res.ravel(), J

    def deterministic_guess(self: Self, T: int | None = None):
        if T is None:
            T = int(self.context.get("constants", {}).get("T", 50))

        y, e = self.__steady_state_vectors__
        v0 = np.concatenate([y, e])[None, :].repeat(T + 1, axis=0)

        for key, value in self.context.get("values", {}).items():
            i = self.symbols["variables"].index(key)
            for a, b in value.items():
                if a <= T:
                    v0[a, i] = b

        return v0

metadata property writable

Convenience accessor for model metadata parsed from source files.

steady_stats property

Convergence statistics and algorithm details from the most recent steady-state calculation.

deterministic_residuals_with_jacobian(v, sparsify=False, continuation='stationary', growth_rate=None, growth_type='geometric', **kwargs)

Stacked-time residuals and Jacobian of the perfect-foresight system.

Generic implementation built on _dynamic_point, one date at a time. Date 0 is pinned to the forced path (initial condition), the trailing exogenous variables follow context["values"] at every date, and the value after the horizon comes from the terminal continuation (see _compute_terminal_continuation).

Source code in src/dyno/model.py
def deterministic_residuals_with_jacobian(
    self: Self,
    v: np.ndarray,
    sparsify: bool = False,
    continuation: str = "stationary",
    growth_rate: Any = None,
    growth_type: str = "geometric",
    **kwargs: Any,
) -> tuple[np.ndarray, Any]:
    """Stacked-time residuals and Jacobian of the perfect-foresight system.

    Generic implementation built on ``_dynamic_point``, one date at a
    time. Date 0 is pinned to the forced path (initial condition), the
    trailing exogenous variables follow ``context["values"]`` at every
    date, and the value after the horizon comes from the terminal
    ``continuation`` (see ``_compute_terminal_continuation``).
    """
    from .dyno_model import (
        _build_dense_jacobian,
        _build_sparse_jacobian,
        _compute_terminal_continuation,
    )

    continuation = kwargs.get("terminal_condition", continuation)
    if continuation in ("static", "steady_static"):
        raise NotImplementedError(
            f"Terminal condition {continuation!r} is only available for DynoModel."
        )

    flat = v.ndim == 1
    variables = list(self.symbols["variables"])
    p = len(variables)
    q = len(self.symbols["endogenous"])
    T = int(np.prod(v.shape) / p - 1)
    v = v.reshape((T + 1, p))

    y, e = self.__steady_state_vectors__
    ss = np.concatenate([y, e])
    v_prev_T = v[-2, :] if T >= 1 else v[-1, :]
    v_next_T, alpha_prev, alpha_curr = _compute_terminal_continuation(
        continuation,
        v_prev_T,
        v[-1, :],
        ss,
        variables,
        growth_rate=growth_rate,
        growth_type=growth_type,
    )
    v_b = np.concatenate([v[0, :][None, :], v[:-1, :]], axis=0)
    v_f = np.concatenate([v[1:, :], v_next_T[None, :]], axis=0)

    v1 = self._forced_path(T)
    N = T + 1
    res = np.zeros((N, p))
    DD = np.zeros((N, p, p, 3))
    for t in range(N):
        r_t, J_t = cast(
            tuple[TVector, TTensor],
            self._dynamic_point(v_b[t], v[t], v_f[t], diff=True),
        )
        res[t, :q] = r_t
        DD[t, :q, :, :] = J_t
    res[:, q:] = v[:, q:] - v1[:, q:]
    for i in range(q, p):
        DD[:, i, i, 1] = 1.0
    res[0, :] = v[0, :] - v1[0, :]  # initial condition

    if not flat:
        return res, DD
    J: Any
    if sparsify:
        J = _build_sparse_jacobian(N, p, DD, {}, alpha_prev, alpha_curr)
    else:
        J = _build_dense_jacobian(N, p, DD, {}, alpha_prev, alpha_curr)
    return res.ravel(), J

equations_with_tags()

Return equations paired with metadata tags and key/value annotations.

Source code in src/dyno/model.py
def equations_with_tags(self: Self) -> list[tuple[int, str, list[str]]]:
    """Return equations paired with metadata tags and key/value annotations."""

    symbolic = getattr(self, "symbolic", None)
    if symbolic is None:
        return []

    pairs: list[tuple[Any, Any]]
    iterator = getattr(symbolic, "iter_equations_with_metadata", None)
    if callable(iterator):
        pairs = list(iterator())
    else:
        equations = list(getattr(symbolic, "equations", []))
        pairs = []
        for eq in equations:
            meta = getattr(eq, "meta", None)
            statement_metadata = (
                getattr(meta, "statement_metadata", {}) if meta is not None else {}
            )
            pairs.append((eq, statement_metadata))

    rows: list[tuple[int, str, list[str]]] = []
    for i, (eq, meta) in enumerate(pairs, start=1):
        eq_text = str(eq)
        try:
            from dyno.dynspec.grammar import str_expression

            eq_text = str_expression(eq)
        except Exception:
            pass

        tags: list[str] = []
        if isinstance(meta, dict):
            raw_tags = meta.get("tags", [])
            if isinstance(raw_tags, list):
                tags = [str(t) for t in raw_tags]
            for k, v in meta.items():
                if k != "tags":
                    tags.append(f"{k}={v}")

        rows.append((i, eq_text, tags))

    return rows

import_model(txt, **kwargs) abstractmethod

Set symbolic attribute from model text description.

Source code in src/dyno/model.py
@abstractmethod
def import_model(self: Self, txt: str, **kwargs: Any) -> None:
    """Set `symbolic` attribute from model text description."""

print_equations_with_tags()

Print all equations with their tags.

Source code in src/dyno/model.py
def print_equations_with_tags(self: Self) -> None:
    """Print all equations with their tags."""

    for i, eq_text, tags in self.equations_with_tags():
        tags_txt = ", ".join(tags) if tags else "-"
        print(f"{i:>3}. {eq_text}  [tags: {tags_txt}]")

recalibrate(**calib)

Return a new model instance with updated parameter/steady-state calibration.

Source code in src/dyno/model.py
def recalibrate(self: Self, **calib: Any) -> Self:
    """Return a new model instance with updated parameter/steady-state calibration."""
    raise NotImplementedError

simulate(T=None, mode='auto', N=1, units=None, solve=None, **args)

Simulate the model.

For deterministic models (or when mode is 'transition' or 'deterministic'), runs the stacked-time perfect foresight solver and returns a TransitionSimulation. For stochastic models, solves the first-order perturbation (or reuses a supplied/cached solution) and delegates to solution.simulate(...).

Parameters:

Name Type Description Default
T int

Simulation horizon. Default is 40 for stochastic, or model-configured/40 for deterministic.

None
mode SimulateMode

Simulation mode ('auto', 'random', 'irf', 'transition', 'deterministic').

'auto'
N int

Number of Monte Carlo simulations (for stochastic random simulation).

1
units UnitsType

Units for output ('level', 'deviation', 'percent', 'log-deviation').

None
solve bool | PerturbationSolution | None

For stochastic models, controls how the model is solved: - None: reuses an already cached solution on the model if present; otherwise solves automatically. - True: forces computing a fresh perturbation solution. - PerturbationSolution: uses the pre-computed solution directly without re-solving. - False: requires that a solution has already been cached on the model.

None
**args Any

Additional options passed to the solver or simulation.

{}
Source code in src/dyno/model.py
def simulate(
    self: Self,
    T: int | None = None,
    mode: SimulateMode = "auto",
    N: int = 1,
    units: UnitsType | None = None,
    solve: bool | PerturbationSolution | None = None,
    **args: Any,
) -> "SimulationResult":
    """Simulate the model.

    For deterministic models (or when ``mode`` is ``'transition'`` or ``'deterministic'``),
    runs the stacked-time perfect foresight solver and returns a ``TransitionSimulation``.
    For stochastic models, solves the first-order perturbation (or reuses a supplied/cached
    solution) and delegates to ``solution.simulate(...)``.

    Parameters
    ----------
    T : int, optional
        Simulation horizon. Default is 40 for stochastic, or model-configured/40 for deterministic.
    mode : SimulateMode, default 'auto'
        Simulation mode ('auto', 'random', 'irf', 'transition', 'deterministic').
    N : int, default 1
        Number of Monte Carlo simulations (for stochastic random simulation).
    units : UnitsType, optional
        Units for output ('level', 'deviation', 'percent', 'log-deviation').
    solve : bool | PerturbationSolution | None, optional
        For stochastic models, controls how the model is solved:
        - ``None``: reuses an already cached solution on the model if present; otherwise solves automatically.
        - ``True``: forces computing a fresh perturbation solution.
        - ``PerturbationSolution``: uses the pre-computed solution directly without re-solving.
        - ``False``: requires that a solution has already been cached on the model.
    **args : Any
        Additional options passed to the solver or simulation.
    """
    self._check_square()
    if self.is_deterministic or mode in ("transition", "deterministic"):
        from .solver import deterministic_solve

        target_units: UnitsType = "level" if units is None else units
        return deterministic_solve(self, T=T, units=target_units, **args)

    from .solver import PerturbationSolution

    sol: PerturbationSolution | None = None
    if isinstance(solve, PerturbationSolution):
        sol = solve
    elif solve is False:
        sol = getattr(self, "_solution", None)
        if sol is None:
            raise ValueError(
                "solve=False was passed, but the model has not been solved yet. "
                "Call model.solve() first or pass solve=True."
            )
    elif solve is True:
        solve_kwargs = {k: v for k, v in args.items() if k in {"method"}}
        sol = self.perturb(**solve_kwargs)
        self._solution = sol
    else:
        # solve is None
        if getattr(self, "_solution", None) is not None:
            sol = self._solution
        else:
            solve_kwargs = {k: v for k, v in args.items() if k in {"method"}}
            sol = self.perturb(**solve_kwargs)
            self._solution = sol

    sim_kwargs = {k: v for k, v in args.items() if k not in {"method"}}
    horizon = 40 if T is None else int(T)
    target_units = "deviation" if units is None else units
    return sol.simulate(T=horizon, mode=mode, N=N, units=target_units, **sim_kwargs)

solve(**args)

Solve the model for its first-order perturbation decision rule.

Only stochastic models with an invariant state-space decision rule can be solved using this method. For deterministic perfect-foresight models, use model.simulate() to compute transition trajectories.

Returns:

Type Description
PerturbationSolution

First-order perturbation solution wrapping the recursive decision rule.

Raises:

Type Description
SystemStructureError

If the model is deterministic, or if the model equations are not square or have unsupported lag/lead timing.

Source code in src/dyno/model.py
def solve(self: Self, **args: Any) -> "PerturbationSolution":
    """Solve the model for its first-order perturbation decision rule.

    Only stochastic models with an invariant state-space decision rule can be solved
    using this method. For deterministic perfect-foresight models, use ``model.simulate()``
    to compute transition trajectories.

    Returns
    -------
    PerturbationSolution
        First-order perturbation solution wrapping the recursive decision rule.

    Raises
    ------
    SystemStructureError
        If the model is deterministic, or if the model equations are not square or
        have unsupported lag/lead timing.
    """
    invalid_shifts = getattr(self, "_invalid_shifts", None)
    if invalid_shifts:
        from .errors import SystemStructureError

        shifts_str = ", ".join(invalid_shifts)
        raise SystemStructureError(
            f"Unsupported timing (only -1, 0, 1 so far): higher-order lead/lag detected ({shifts_str}). "
            "Dyno solvers currently support shifts in [-1, 1]."
        )

    self._check_square()
    if self.is_deterministic:
        from .errors import SystemStructureError

        raise SystemStructureError(
            "Deterministic models do not have a perturbation decision rule and cannot be solved with `model.solve()`. "
            "Use `model.simulate()` instead to compute the perfect-foresight transition trajectory."
        )
    sol = self.perturb(**args)
    self._solution = sol
    return sol

variants(*args, **kwargs)

Create a VariantCollection holding differently calibrated versions of this model.

Examples:

>>> var = model.variants(a=[2, 3, 4])
>>> sims = model.variants(a=[2, 3, 4]).solve().simulate()
>>> fig = sims.plot()
Source code in src/dyno/model.py
def variants(self: Self, *args: Any, **kwargs: Any) -> "VariantCollection[Self]":
    """Create a ``VariantCollection`` holding differently calibrated versions of this model.

    Examples
    --------
    >>> var = model.variants(a=[2, 3, 4])
    >>> sims = model.variants(a=[2, 3, 4]).solve().simulate()
    >>> fig = sims.plot()
    """
    from .variants import VariantCollection, _expand_variant_specs

    specs, labels = _expand_variant_specs(args, kwargs)
    models = [self.recalibrate(**spec) for spec in specs]
    return VariantCollection(models, specs=specs, labels=labels)

Bases: AbstractModel

Source code in src/dyno/dyno_model.py
 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
class DynoModel(AbstractModel):

    def _normalize_run_commands(self: Self) -> list[dict[str, Any]]:
        raw = self.metadata.get("run")
        if raw is None:
            raw = self.metadata.get("dynare_commands", [])

        items: list[Any]
        if isinstance(raw, str):
            items = [raw]
        elif isinstance(raw, dict):
            items = [raw]
        elif isinstance(raw, list):
            items = raw
        else:
            raise TypeError("model.metadata['run'] must be a string, dict or a list")

        commands: list[dict[str, Any]] = []
        for item in items:
            if isinstance(item, str):
                mute = False
                if item.endswith(";"):
                    mute = True
                    item = item[:-1].rstrip()
                if ";" in item:
                    raise ValueError(
                        f"Invalid run command '{item}': unexpected semicolon."
                    )
                commands.append({"command": item, "options": {}, "mute": mute})
            elif isinstance(item, dict):
                mute = bool(item.get("mute", False))
                if "command" in item:
                    command = item.get("command")
                    if not isinstance(command, str):
                        raise TypeError(
                            "run command dictionaries must define a string 'command'"
                        )
                    if command.endswith(";"):
                        mute = True
                        command = command[:-1].rstrip()
                    if ";" in command:
                        raise ValueError(
                            f"Invalid run command '{command}': unexpected semicolon."
                        )
                    options = item.get("options", {})
                    if not isinstance(options, dict):
                        raise TypeError("run command 'options' must be a dictionary")
                    commands.append(
                        {"command": command, "options": options, "mute": mute}
                    )
                else:
                    non_mute_keys = [k for k in item if k != "mute"]
                    if len(non_mute_keys) == 1:
                        command = non_mute_keys[0]
                        options = item[command]
                        if not isinstance(command, str):
                            raise TypeError("run command keys must be strings")
                        if ";" in command:
                            raise ValueError(
                                f"Invalid run command '{command}': semicolons are not permitted in command names."
                            )
                        if options is None:
                            options = {}
                        if not isinstance(options, dict):
                            raise TypeError(
                                "compact run command options must be a dictionary or null"
                            )
                        commands.append(
                            {"command": command, "options": options, "mute": mute}
                        )
                    else:
                        raise TypeError(
                            "run command dictionaries must use either {'command': ...} or a single-key form like {'simul': {...}}"
                        )
            else:
                raise TypeError("run commands must be strings or dictionaries")

        # File-wide ``@variants`` metadata replaces the model by a collection
        # of models: equivalent to a leading ``variants`` command, so that the
        # whole pipeline runs on each variant.
        variants_spec = self.metadata.get("variants")
        if variants_spec is not None and not any(
            str(c["command"]).lower() == "variants" for c in commands
        ):
            if not isinstance(variants_spec, dict):
                raise TypeError(
                    "model.metadata['variants'] must be a mapping of parameter "
                    "names to lists of values"
                )
            commands.insert(
                0,
                {"command": "variants", "options": dict(variants_spec), "mute": False},
            )

        return commands

    def run(self: Self, default_pipeline: bool = False) -> "RunResults":
        """Execute the model's ``@run:`` pipeline and collect the results.

        Commands are read from ``model.metadata["run"]`` (populated by
        ``@run:`` directives in a ``.dyno`` file, a ``run:`` list in a YAML
        model, or the computing commands of a ``.mod`` file) and executed in
        order. The ``@run:`` command format is stable.

        Each command can be written as:

        - a bare name: ``@run: steady``
        - a single-key mapping with options: ``@run: simul: {T: 40}``
          (``null`` options are allowed, e.g. ``steady: null``)
        - an explicit mapping: ``{command: simul, options: {T: 40}}``

        A trailing ``;`` at the end of the command line (``@run: solve;``
        or ``@run: simul: {T: 40};``) mutes that command's output in reports.
        Semicolons are only permitted at the end of the line.

        Supported commands:

        - ``steady``: solve for the steady state (``model.steady(**options)``)
        - ``resid``: compute steady-state residuals
        - ``check``: compute residuals and check Blanchard-Kahn conditions
        - ``solve`` / ``perturb``: compute the first-order solution
        - ``simul`` / ``simulate``: simulate the model (perfect-foresight path
          for deterministic models)
        - ``analyze`` / ``stoch_simul``: solve, compute moments and IRFs
        - ``plot``: plot the current simulation
        - ``variants``: run the remaining commands on each calibration variant
          (``model.variants(**options)``)

        The file-wide metadata ``@variants: {alpha: [1, 2]}`` is equivalent to
        a leading ``variants`` command: it replaces the model by a collection
        of models and the whole pipeline runs on each of them.

        Parameters
        ----------
        default_pipeline : bool, optional
            If True and the model defines no commands, run ``resid`` followed
            by ``analyze`` (``T=40``). By default False.

        Returns
        -------
        RunResults
            Container with the final model, residuals, solution, eigenvalues,
            moments, simulation and figure produced by the pipeline (a
            collection of results when ``variants`` is used).

        Raises
        ------
        NotImplementedError
            If a command is not supported.
        """
        from .report import RunResults

        commands = self._normalize_run_commands()
        model = self
        results = RunResults(model=model)
        results._from_pipeline = True
        results.muted_commands = {
            str(c["command"]).lower() for c in commands if c.get("mute")
        }

        invalid_shifts = getattr(self, "_invalid_shifts", None)
        if invalid_shifts:
            from .errors import SystemStructureError

            shifts_str = ", ".join(invalid_shifts)
            raise SystemStructureError(
                f"Unsupported timing (only -1, 0, 1 so far): higher-order lead/lag detected ({shifts_str}). "
                "Dyno solvers currently support shifts in [-1, 1]."
            )

        if not commands and default_pipeline:
            commands = [
                {"command": "resid", "options": {}},
                {"command": "analyze", "options": {"T": 40}},
            ]

        from .errors import BlanchardKahnError

        bk_failed = False

        def _solve_or_record(solve_options: dict[str, Any]) -> Any:
            """Solve the model, or record a Blanchard-Kahn failure and return None.

            Without a stable solution, later commands that need one are skipped
            instead of aborting the whole report.
            """
            nonlocal bk_failed
            if bk_failed:
                return None
            try:
                return model.solve(**solve_options)
            except BlanchardKahnError as e:
                bk_failed = True
                results.eigenvalues = e.evs
                results.add_error(str(e))
                return None

        for idx, cmd in enumerate(commands):
            name = str(cmd["command"]).lower()
            options = cmd.get("options", {})

            if name == "variants":
                model_variants = model.variants(**options)
                remaining = commands[idx + 1 :]
                for m in model_variants:
                    m.metadata = dict(m.metadata)
                    m.metadata.pop("variants", None)  # already expanded
                    m.metadata["run"] = remaining
                return model_variants.run(default_pipeline=default_pipeline)  # type: ignore[return-value]
            elif name == "steady":
                from .errors import SteadyStateError

                try:
                    model = model.steady(**options)
                    results.model = model
                    results.steady_stats = getattr(model, "steady_stats", None)
                except SteadyStateError as e:
                    results.residuals = e.residuals
                    results.steady_stats = getattr(e, "steady_stats", None)
                    results.add_warning(str(e))
            elif name == "resid":
                results.residuals = model.residuals
            elif name == "check":
                from .errors import SteadyStateError

                results.residuals = model.residuals
                check_options = {"compute_eigenvalues": True, **options}
                try:
                    model = model.check(**check_options)
                    results.model = model
                    results.eigenvalues = getattr(model, "_eigenvalues", None)
                except SteadyStateError as e:
                    results.residuals = e.residuals
                    results.add_warning(str(e))
            elif name in {"solve", "perturb"}:
                if model.is_deterministic:
                    results.add_warning(
                        "Command 'solve' is not applicable to deterministic models. "
                        "Use 'simulate' instead to compute the transition trajectory."
                    )
                    continue
                solution: PerturbationSolution | None = _solve_or_record(options)
                if solution is None:
                    continue
                results.solution = solution
                results.eigenvalues = getattr(solution, "evs", None)
            elif name in {"simul", "simulate"}:
                mode = options.get("mode", "random" if name == "simul" else "auto")
                if model.is_deterministic or mode in {"transition", "deterministic"}:
                    from .solver import deterministic_solve

                    det_opts = {k: v for k, v in options.items() if k != "mode"}
                    sim = deterministic_solve(model, **det_opts)
                    results.simulation = sim
                    if hasattr(sim, "attrs") and not sim.attrs.get("converged", True):
                        results.add_warning(
                            f"Deterministic / perfect foresight simulation did not converge after "
                            f"{sim.attrs.get('iterations', '?')} iterations "
                            f"(maximum residual: {sim.attrs.get('residual', float('nan')):.2e}). "
                            "The computed solution is incorrect."
                        )
                else:
                    solve_options = {
                        k: v for k, v in options.items() if k in {"method"}
                    }
                    solution = results.solution
                    if solution is None or not hasattr(solution, "X"):
                        solution = _solve_or_record(solve_options)
                        if solution is None:
                            continue
                        results.solution = solution
                    results.eigenvalues = getattr(solution, "evs", None)
                    horizon = int(
                        options.get("T", options.get("irf", options.get("periods", 40)))
                    )
                    units = options.get("units", options.get("type", "deviation"))
                    n_draws = int(options.get("N", 1))
                    sim_options = {
                        k: v
                        for k, v in options.items()
                        if k in {"shocks", "initial_states", "rng"}
                    }
                    results.simulation = solution.simulate(
                        T=horizon,
                        mode=mode,
                        N=n_draws,
                        units=units,
                        **sim_options,
                    )
            elif name in {"analyze", "stoch_simul"}:
                from .plots import plot_simulation

                plot_variables = options.get("variables", options.get("vars"))
                plot_options: dict[str, Any] = {"engine": "altair"}
                if plot_variables:
                    plot_options["variables"] = plot_variables
                if name == "stoch_simul":
                    # Dynare semantics: ``periods`` is the length of the
                    # stochastic simulation, not the IRF horizon.
                    horizon = int(options.get("T", options.get("irf", 40)))
                    show_graph = not options.get("nograph", False) and horizon > 0
                else:
                    horizon = int(
                        options.get("T", options.get("irf", options.get("periods", 40)))
                    )
                    show_graph = True

                if model.is_deterministic:
                    from .solver import deterministic_solve

                    det_opts = {
                        k: v
                        for k, v in options.items()
                        if k
                        not in {
                            "irf",
                            "periods",
                            "type",
                            "variables",
                            "vars",
                            "nograph",
                        }
                    }
                    det_opts["T"] = horizon
                    sim = deterministic_solve(model, **det_opts)
                    results.simulation = sim
                    if hasattr(sim, "attrs") and not sim.attrs.get("converged", True):
                        results.add_warning(
                            f"Deterministic / perfect foresight simulation did not converge after "
                            f"{sim.attrs.get('iterations', '?')} iterations "
                            f"(maximum residual: {sim.attrs.get('residual', float('nan')):.2e}). "
                            "The computed solution is incorrect."
                        )
                else:
                    solve_options = {
                        k: v for k, v in options.items() if k in {"method"}
                    }
                    solution = results.solution
                    if solution is None or not hasattr(solution, "X"):
                        solution = _solve_or_record(solve_options)
                        if solution is None:
                            continue
                        results.solution = solution
                    results.eigenvalues = getattr(solution, "evs", None)
                    irf_type = options.get("units", options.get("type", "deviation"))
                    results.moments = solution.moments()[1]
                    if horizon > 0:
                        results.simulation = solution.irfs(type=irf_type, T=horizon)
                if show_graph and results.simulation is not None:
                    results._plot_options = plot_options
                    results.figure = plot_simulation(
                        results.simulation, engine="altair", variables=plot_variables
                    )
            elif name == "plot":
                if results.simulation is not None:
                    from .plots import plot_simulation

                    plot_options = dict(options)
                    if "vars" in plot_options and "variables" not in plot_options:
                        plot_options["variables"] = plot_options.pop("vars")
                    if "type" in plot_options and "units" not in plot_options:
                        plot_options["units"] = plot_options.pop("type")
                    engine = plot_options.get("engine", "altair")
                    plot_opts = {k: v for k, v in plot_options.items() if k != "engine"}
                    results._plot_options = dict(plot_options)  # type: ignore[attr-defined]
                    results.figure = plot_simulation(
                        results.simulation, engine=engine, **plot_opts
                    )
            else:
                raise NotImplementedError(f"Unsupported DynoModel.run command: {name}")

        # Add line-level warnings for non-zero residuals
        r = results.residuals
        if r is not None and r.size > 0 and abs(r).max() >= 1e-6:
            inds = np.where(abs(r) >= 1e-6)[0]
            for i in inds:
                tree = model.symbolic.equations[i]
                results.add_warning(str(r[i]), line=tree.meta.line)

        results.finish()
        return results

    def _constants_used_in_equations(self: Self) -> set[str]:
        names: set[str] = set()
        for eq in self.symbolic.equations:
            for subtree in eq.iter_subtrees_topdown():
                if subtree.data == "constant":
                    names.add(str(subtree.children[0].children[0]))
        return names

    def _variables_used_in_equations(self: Self) -> set[str]:
        names: set[str] = set()
        for eq in getattr(self.symbolic, "equations", []):
            if hasattr(eq, "iter_subtrees_topdown"):
                for subtree in eq.iter_subtrees_topdown():
                    if getattr(subtree, "data", None) == "variable":
                        if subtree.children and hasattr(
                            subtree.children[0], "children"
                        ):
                            names.add(str(subtree.children[0].children[0]))
        return names

    def __init__(
        self: Self,
        filename: str | os.PathLike[str] | None = None,
        txt: str | None = None,
        yaml: str | None = None,
        strict: bool = False,
        preprocess: bool = True,
        include_paths: list[str] | None = None,
        defines: dict[str, Any] | None = None,
        **kwargs,
    ) -> None:
        if yaml is not None:
            if txt is not None:
                raise ValueError("Pass either `txt` or `yaml`, not both.")
            txt = yaml
            if filename is None:
                filename = "*anonymous*.yaml"

        super().__init__(
            filename=filename,
            txt=txt,
            strict=strict,
            preprocess=preprocess,
            include_paths=include_paths,
            defines=defines,
            **kwargs,
        )
        self._warn_suspicious_variables()

    def _warn_suspicious_variables(self: Self) -> None:
        """Warn about endogenous variables that are likely typos.

        In Dyno syntax any name followed by ``[t]`` becomes a variable, so a
        misspelled name silently adds an endogenous variable. Flag those that
        appear in a single equation and have no steady-state declaration.
        Dynare files are skipped: their variables are declared explicitly.
        """
        if not isinstance(self.symbolic, DynoFile):
            return
        hints = self._suspicious_endogenous(require_single_occurrence=True)
        if hints:
            from .errors import UndefinedSymbolWarning

            warnings.warn(
                "Possible typo in variable name:\n"
                + "\n".join(f"  {h}" for h in hints),
                UndefinedSymbolWarning,
                stacklevel=3,
            )

    def import_model(
        self: Self,
        txt: str,
        preprocess: bool = True,
        include_paths: list[str] | None = None,
        defines: dict[str, Any] | None = None,
        **kwargs,
    ) -> None:

        try:
            if self.filename.endswith(".mod"):
                if preprocess:
                    from dyno.dynare.macro import expand_macro

                    txt = expand_macro(
                        txt,
                        filepath=(
                            self.filename if not self.filename.startswith("*") else None
                        ),
                        include_paths=include_paths,
                        defines=defines,
                        only_if_needed=True,
                    )
                self.symbolic = LModFile(content=txt, filename=self.filename)
            elif self.filename.endswith(".yaml") or self.filename.endswith(".yml"):
                data = yaml.safe_load(txt)
                if not isinstance(data, dict):
                    raise ValueError(
                        "YAML model input must be a mapping with a `model` key."
                    )

                model_txt = data.get("model")
                if not isinstance(model_txt, str):
                    raise ValueError(
                        "YAML model input must provide a string `model` block."
                    )

                self.symbolic = DynoFile(content=model_txt, filename=self.filename)

                yaml_metadata = {k: v for (k, v) in data.items() if k != "model"}
                merged_metadata = yaml_metadata | self.symbolic.metadata
                self.symbolic.metadata = merged_metadata
            elif self.filename.endswith(".dyno"):
                self.symbolic = DynoFile(content=txt, filename=self.filename)
            else:
                self.symbolic = DynoFile(content=txt, filename=self.filename)
        except UnexpectedInput as e:
            raise LARKParserError(e, txt) from e

    def _set_context(self: Self) -> None:
        context = self.symbolic.context.copy()

        constants = context.setdefault("constants", {})
        steady_states = context.setdefault("steady_states", {})
        variables = context.setdefault("variables", {})

        # Ensure constants used in equations are explicit in context, even if
        # they were never assigned in declarations.
        unassigned_constants = []
        for name in self._constants_used_in_equations():
            if name not in constants:
                constants[name] = math.nan
                unassigned_constants.append(name)
            elif isinstance(constants[name], float) and math.isnan(constants[name]):
                unassigned_constants.append(name)

        if unassigned_constants:
            from .errors import UndefinedSymbolError, UndefinedSymbolWarning

            names_str = ", ".join(sorted(set(unassigned_constants)))
            msg = f"Undefined parameter(s) used in equation definitions: {names_str}"
            if getattr(self, "strict", False):
                raise UndefinedSymbolError(msg)
            else:
                warnings.warn(
                    f"{msg}. Defaulting to NaN.",
                    UndefinedSymbolWarning,
                    stacklevel=2,
                )

        # Detect higher-order leads/lags in equations
        invalid_shifts = []
        for eq in getattr(self.symbolic, "equations", []):
            for subtree in eq.iter_subtrees_topdown():
                if subtree.data == "variable":
                    shift = int(subtree.children[2].children[0])
                    if abs(shift) > 1:
                        vname = str(subtree.children[0].children[0])
                        invalid_shifts.append(f"{vname}[t{shift:+d}]")
        self._invalid_shifts = invalid_shifts
        if invalid_shifts:
            from .errors import SystemStructureError, UndefinedSymbolWarning

            shifts_str = ", ".join(invalid_shifts)
            msg = f"Unsupported timing (only -1, 0, 1 so far): higher-order lead/lag detected ({shifts_str}). Dyno solvers currently support shifts in [-1, 1]."
            if getattr(self, "strict", False):
                raise SystemStructureError(msg)
            else:
                warnings.warn(msg, UndefinedSymbolWarning, stacklevel=2)

        # Ensure every referenced variable has a steady-state entry.
        for name in variables.keys():
            steady_states.setdefault(name, math.nan)

        self.context = cast(ModelContext, context)

    def recalibrate(self: Self, **calib):
        m = self.copy()
        prev_metadata = copy.deepcopy(getattr(self.symbolic, "metadata", {}))
        previous = getattr(self, "_calibration_overrides", {})
        merged = previous | calib
        m.symbolic.context = {}
        m.symbolic.process_assignments(**merged)
        m.symbolic.metadata = prev_metadata | getattr(m.symbolic, "metadata", {})
        m._set_context()
        m._set_exogenous()
        m._calibration_overrides = merged
        return m

    @property
    def equations(self):

        return [str_expression(eq) for eq in self.symbolic.equations]

    def latex_equations(self):

        return self.symbolic.latex_equations()

    def compute_residuals(self, y2, y1, y0, e):

        endogenous = self.symbols["endogenous"]
        exogenous = self.symbols["exogenous"]

        cc = copy.deepcopy(self.symbolic.context)

        for i, name in enumerate(endogenous):
            cc["variables"][name] = {
                -1: y0[i],
                0: y1[i],
                1: y2[i],
            }

        for i, name in enumerate(exogenous):
            exo_val = (
                e[i]
                if (i < len(e) and not math.isnan(e[i]))
                else cc["steady_states"].get(name, 0.0)
            )
            if isinstance(exo_val, float) and math.isnan(exo_val):
                exo_val = 0.0
            cc["variables"][name] = {
                -1: exo_val,
                0: exo_val,
                1: exo_val,
            }

        from dyno.dynspec.analyze import EquationsEvaluator

        E = EquationsEvaluator(cc)

        results = [E.visit(eq) for eq in self.symbolic.equations]

        r = np.array([float(el) for el in results])

        return r

    def compute_jacobians(
        self, y2, y1, y0, e
    ) -> tuple[TVector, TMatrix, TMatrix, TMatrix, TMatrix]:

        from dyno.dynspec.autodiff import DNumber as DN

        endogenous = self.symbols["endogenous"]
        exogenous = self.symbols["exogenous"]

        cc = copy.deepcopy(self.symbolic.context)

        for i, name in enumerate(endogenous):
            cc["variables"][name] = {
                -1: DN(y0[i], {(name, -1): 1}),
                0: DN(y1[i], {(name, 0): 1}),
                1: DN(y2[i], {(name, 1): 1}),
            }

        for i, name in enumerate(exogenous):
            exo_val = (
                e[i]
                if (i < len(e) and not math.isnan(e[i]))
                else cc["steady_states"].get(name, 0.0)
            )
            if isinstance(exo_val, float) and math.isnan(exo_val):
                exo_val = 0.0
            cc["variables"][name] = {
                -1: DN(exo_val, {(name, 0): 1}),
                0: DN(exo_val, {(name, 0): 1}),
                1: DN(exo_val, {(name, 0): 1}),
            }

        from dyno.dynspec.analyze import EquationsEvaluator

        E = EquationsEvaluator(cc)
        results = [E.visit(eq) for eq in self.symbolic.equations]

        neq = len(results)
        nv = len(endogenous)
        ne = len(exogenous)

        r = np.array([el.value for el in results])
        A = np.zeros((neq, nv))
        B = np.zeros((neq, nv))
        C = np.zeros((neq, nv))
        J = [A, B, C]
        D = np.zeros((neq, ne))

        for n, eq in enumerate(results):
            for (name, shift), v in eq.derivatives.items():
                if name in endogenous:
                    i = endogenous.index(name)
                    J[1 - shift][n, i] = v
                elif name in exogenous:
                    i = exogenous.index(name)
                    D[n, i] = v

        return r, A, B, C, D

    def _dynamic_point(self, v_prev, v_curr, v_next, diff=False):
        from dyno.dynspec.analyze import EquationsEvaluator
        from dyno.dynspec.autodiff import DNumber as DN

        variables = self.symbols["variables"]
        cc = copy.deepcopy(self.symbolic.context)
        for i, name in enumerate(variables):
            if diff:
                cc["variables"][name] = {
                    -1: DN(v_prev[i], {(name, -1): 1}),
                    0: DN(v_curr[i], {(name, 0): 1}),
                    1: DN(v_next[i], {(name, 1): 1}),
                }
            else:
                cc["variables"][name] = {-1: v_prev[i], 0: v_curr[i], 1: v_next[i]}
        results = [EquationsEvaluator(cc).visit(eq) for eq in self.symbolic.equations]
        if not diff:
            return np.array([float(el) for el in results])
        r = np.array([el.value for el in results])
        J = np.zeros((len(results), len(variables), 3))
        for n, eq in enumerate(results):
            for (name, shift), value in eq.derivatives.items():
                J[n, variables.index(name), shift + 1] = value
        return r, J

    def deterministic_residuals(
        model,
        v,
        jac=False,
        continuation="stationary",
        growth_rate=None,
        growth_type="geometric",
        **kwargs,
    ):

        continuation = kwargs.get("terminal_condition", continuation)

        if jac:
            return model.deterministic_residuals_with_jacobian(
                v,
                continuation=continuation,
                growth_rate=growth_rate,
                growth_type=growth_type,
                **kwargs,
            )

        flat = v.ndim == 1

        p = len(model.symbols["variables"])
        T = int(np.prod(v.shape) / p - 1)

        v = v.reshape((T + 1, p))

        # For t = 0 to T: forward/backward indexing
        v_b = np.concatenate([v[0, :][None, :], v[:-1, :]], axis=0)

        y, e = model.__steady_state_vectors__
        ss = np.concatenate([y, e])

        v_prev = v[-2, :] if T >= 1 else v[-1, :]
        v_curr = v[-1, :]

        v_next, _, _ = _compute_terminal_continuation(
            continuation,
            v_prev,
            v_curr,
            ss,
            model.symbols["variables"],
            growth_rate=growth_rate,
            growth_type=growth_type,
        )

        v_f = np.concatenate([v[1:, :], v_next[None, :]], axis=0)

        cc = copy.deepcopy(model.symbolic.context)
        for i, name in enumerate(model.symbols["variables"]):
            cc["variables"][name] = {-1: v_b[:, i], 0: v[:, i], 1: v_f[:, i]}

        from dyno.dynspec.analyze import EquationsEvaluator

        E = EquationsEvaluator(cc)

        results = [E.visit(eq) for eq in model.symbolic.equations]

        # number of variables not pinned down by dynamic equations
        n_exo = len(model.symbols["variables"]) - len(results)

        v1 = ss[None, :].repeat(T + 1, axis=0)
        for key, value in model.symbolic.context["values"].items():
            i = model.symbols["variables"].index(key)
            for a, b in value.items():
                if a <= T:
                    v1[a, i] = b

        v1_exo = v1[:, -n_exo:].copy()
        diff_exo = v[:, -n_exo:] - v1_exo

        res = np.column_stack(results + [diff_exo])

        res[0, :] = v[0, :] - v1[0, :]  # initial condition

        if continuation in ("static", "steady_static") and T >= 1:
            # Legacy static terminal condition: f(v_T, v_T, v_T)
            cc_T = copy.deepcopy(model.symbolic.context)
            for i, name in enumerate(model.symbols["variables"]):
                cc_T["variables"][name] = {-1: v[T, i], 0: v[T, i], 1: v[T, i]}
            E_T = EquationsEvaluator(cc_T)
            res_T = [E_T.visit(eq) for eq in model.symbolic.equations]
            res[T, : len(res_T)] = res_T
            res[T, len(res_T) :] = v[T, -n_exo:] - v1_exo[T, :]

        if flat:
            return res.ravel()
        else:
            return res

    def deterministic_residuals_with_jacobian(
        model,
        v,
        sparsify=False,
        continuation="stationary",
        growth_rate=None,
        growth_type="geometric",
        **kwargs,
    ):

        from dyno.dynspec.autodiff import DNumber

        continuation = kwargs.get("terminal_condition", continuation)

        flat = v.ndim == 1

        p = len(model.symbols["variables"])
        T = int(np.prod(v.shape) / p - 1)

        v = v.reshape((T + 1, p))

        v_b = np.concatenate([v[0, :][None, :], v[:-1, :]], axis=0)

        y, e = model.__steady_state_vectors__
        ss = np.concatenate([y, e])

        v_prev = v[-2, :] if T >= 1 else v[-1, :]
        v_curr = v[-1, :]

        v_next, alpha_prev, alpha_curr = _compute_terminal_continuation(
            continuation,
            v_prev,
            v_curr,
            ss,
            model.symbols["variables"],
            growth_rate=growth_rate,
            growth_type=growth_type,
        )

        v_f = np.concatenate([v[1:, :], v_next[None, :]], axis=0)

        context = copy.deepcopy(model.symbolic.context)
        for i, name in enumerate(model.symbols["variables"]):
            context["variables"][name] = {
                -1: DNumber(v_b[:, i], {(name, -1): 1.0}),
                0: DNumber(v[:, i], {(name, 0): 1.0}),
                1: DNumber(v_f[:, i], {(name, 1): 1.0}),
            }

        from dyno.dynspec.analyze import EquationsEvaluator

        E = EquationsEvaluator(context)

        results = [E.visit(eq) for eq in model.symbolic.equations]

        n_exo = len(model.symbols["variables"]) - len(results)

        v1 = ss[None, :].repeat(T + 1, axis=0)
        for key, value in model.symbolic.context["values"].items():
            i = model.symbols["variables"].index(key)
            for a, b in value.items():
                if a <= T:
                    v1[a, i] = b

        exo = v1[:, -n_exo:].copy()

        res = np.column_stack([e.value for e in results] + [v[:, -n_exo:] - exo])

        res[0, :] = v[0, :] - v1[0, :]  # initial condition

        N = v.shape[0]

        q = len(model.equations)

        D = np.zeros((N, q, p, 3))

        for i_q in range(q):
            for k, deriv in results[i_q].derivatives.items():
                s, t = k  # symbol, time
                i_var = model.symbols["variables"].index(s)
                D[:, i_q, i_var, t + 1] = deriv

        # add exogenous equations
        DD = np.zeros((N, p, p, 3))
        DD[:, :q, :, :] = D
        for i in range(q, p):
            DD[:, i, i, 1] = 1.0

        terminal_derivatives = {}

        if continuation in ("static", "steady_static") and T >= 1:
            context_T = copy.deepcopy(model.symbolic.context)
            for i, name in enumerate(model.symbols["variables"]):
                context_T["variables"][name] = {
                    -1: DNumber(v[T, i], {(name, -1): 1.0}),
                    0: DNumber(v[T, i], {(name, 0): 1.0}),
                    1: DNumber(v[T, i], {(name, 1): 1.0}),
                }
            E_T = EquationsEvaluator(context_T)
            results_T = [E_T.visit(eq) for eq in model.symbolic.equations]
            res[T, :q] = [e.value for e in results_T]
            res[T, q:] = v[T, -n_exo:] - exo[T, :]

            deriv_T_T = np.zeros((p, p))
            for i_q in range(q):
                for k, deriv in results_T[i_q].derivatives.items():
                    s, t = k
                    i_var = model.symbols["variables"].index(s)
                    deriv_T_T[i_q, i_var] += deriv
            for i in range(q, p):
                deriv_T_T[i, i] = 1.0
            terminal_derivatives[(T, T)] = deriv_T_T

            DD[T, :, :, :] = 0.0

        if not flat:
            return res, DD
        else:
            if sparsify:
                J = _build_sparse_jacobian(
                    N, p, DD, terminal_derivatives, alpha_prev, alpha_curr
                )
            else:
                J = _build_dense_jacobian(
                    N, p, DD, terminal_derivatives, alpha_prev, alpha_curr
                )

            return res.ravel(), J

run(default_pipeline=False)

Execute the model's @run: pipeline and collect the results.

Commands are read from model.metadata["run"] (populated by @run: directives in a .dyno file, a run: list in a YAML model, or the computing commands of a .mod file) and executed in order. The @run: command format is stable.

Each command can be written as:

  • a bare name: @run: steady
  • a single-key mapping with options: @run: simul: {T: 40} (null options are allowed, e.g. steady: null)
  • an explicit mapping: {command: simul, options: {T: 40}}

A trailing ; at the end of the command line (@run: solve; or @run: simul: {T: 40};) mutes that command's output in reports. Semicolons are only permitted at the end of the line.

Supported commands:

  • steady: solve for the steady state (model.steady(**options))
  • resid: compute steady-state residuals
  • check: compute residuals and check Blanchard-Kahn conditions
  • solve / perturb: compute the first-order solution
  • simul / simulate: simulate the model (perfect-foresight path for deterministic models)
  • analyze / stoch_simul: solve, compute moments and IRFs
  • plot: plot the current simulation
  • variants: run the remaining commands on each calibration variant (model.variants(**options))

The file-wide metadata @variants: {alpha: [1, 2]} is equivalent to a leading variants command: it replaces the model by a collection of models and the whole pipeline runs on each of them.

Parameters:

Name Type Description Default
default_pipeline bool

If True and the model defines no commands, run resid followed by analyze (T=40). By default False.

False

Returns:

Type Description
RunResults

Container with the final model, residuals, solution, eigenvalues, moments, simulation and figure produced by the pipeline (a collection of results when variants is used).

Raises:

Type Description
NotImplementedError

If a command is not supported.

Source code in src/dyno/dyno_model.py
def run(self: Self, default_pipeline: bool = False) -> "RunResults":
    """Execute the model's ``@run:`` pipeline and collect the results.

    Commands are read from ``model.metadata["run"]`` (populated by
    ``@run:`` directives in a ``.dyno`` file, a ``run:`` list in a YAML
    model, or the computing commands of a ``.mod`` file) and executed in
    order. The ``@run:`` command format is stable.

    Each command can be written as:

    - a bare name: ``@run: steady``
    - a single-key mapping with options: ``@run: simul: {T: 40}``
      (``null`` options are allowed, e.g. ``steady: null``)
    - an explicit mapping: ``{command: simul, options: {T: 40}}``

    A trailing ``;`` at the end of the command line (``@run: solve;``
    or ``@run: simul: {T: 40};``) mutes that command's output in reports.
    Semicolons are only permitted at the end of the line.

    Supported commands:

    - ``steady``: solve for the steady state (``model.steady(**options)``)
    - ``resid``: compute steady-state residuals
    - ``check``: compute residuals and check Blanchard-Kahn conditions
    - ``solve`` / ``perturb``: compute the first-order solution
    - ``simul`` / ``simulate``: simulate the model (perfect-foresight path
      for deterministic models)
    - ``analyze`` / ``stoch_simul``: solve, compute moments and IRFs
    - ``plot``: plot the current simulation
    - ``variants``: run the remaining commands on each calibration variant
      (``model.variants(**options)``)

    The file-wide metadata ``@variants: {alpha: [1, 2]}`` is equivalent to
    a leading ``variants`` command: it replaces the model by a collection
    of models and the whole pipeline runs on each of them.

    Parameters
    ----------
    default_pipeline : bool, optional
        If True and the model defines no commands, run ``resid`` followed
        by ``analyze`` (``T=40``). By default False.

    Returns
    -------
    RunResults
        Container with the final model, residuals, solution, eigenvalues,
        moments, simulation and figure produced by the pipeline (a
        collection of results when ``variants`` is used).

    Raises
    ------
    NotImplementedError
        If a command is not supported.
    """
    from .report import RunResults

    commands = self._normalize_run_commands()
    model = self
    results = RunResults(model=model)
    results._from_pipeline = True
    results.muted_commands = {
        str(c["command"]).lower() for c in commands if c.get("mute")
    }

    invalid_shifts = getattr(self, "_invalid_shifts", None)
    if invalid_shifts:
        from .errors import SystemStructureError

        shifts_str = ", ".join(invalid_shifts)
        raise SystemStructureError(
            f"Unsupported timing (only -1, 0, 1 so far): higher-order lead/lag detected ({shifts_str}). "
            "Dyno solvers currently support shifts in [-1, 1]."
        )

    if not commands and default_pipeline:
        commands = [
            {"command": "resid", "options": {}},
            {"command": "analyze", "options": {"T": 40}},
        ]

    from .errors import BlanchardKahnError

    bk_failed = False

    def _solve_or_record(solve_options: dict[str, Any]) -> Any:
        """Solve the model, or record a Blanchard-Kahn failure and return None.

        Without a stable solution, later commands that need one are skipped
        instead of aborting the whole report.
        """
        nonlocal bk_failed
        if bk_failed:
            return None
        try:
            return model.solve(**solve_options)
        except BlanchardKahnError as e:
            bk_failed = True
            results.eigenvalues = e.evs
            results.add_error(str(e))
            return None

    for idx, cmd in enumerate(commands):
        name = str(cmd["command"]).lower()
        options = cmd.get("options", {})

        if name == "variants":
            model_variants = model.variants(**options)
            remaining = commands[idx + 1 :]
            for m in model_variants:
                m.metadata = dict(m.metadata)
                m.metadata.pop("variants", None)  # already expanded
                m.metadata["run"] = remaining
            return model_variants.run(default_pipeline=default_pipeline)  # type: ignore[return-value]
        elif name == "steady":
            from .errors import SteadyStateError

            try:
                model = model.steady(**options)
                results.model = model
                results.steady_stats = getattr(model, "steady_stats", None)
            except SteadyStateError as e:
                results.residuals = e.residuals
                results.steady_stats = getattr(e, "steady_stats", None)
                results.add_warning(str(e))
        elif name == "resid":
            results.residuals = model.residuals
        elif name == "check":
            from .errors import SteadyStateError

            results.residuals = model.residuals
            check_options = {"compute_eigenvalues": True, **options}
            try:
                model = model.check(**check_options)
                results.model = model
                results.eigenvalues = getattr(model, "_eigenvalues", None)
            except SteadyStateError as e:
                results.residuals = e.residuals
                results.add_warning(str(e))
        elif name in {"solve", "perturb"}:
            if model.is_deterministic:
                results.add_warning(
                    "Command 'solve' is not applicable to deterministic models. "
                    "Use 'simulate' instead to compute the transition trajectory."
                )
                continue
            solution: PerturbationSolution | None = _solve_or_record(options)
            if solution is None:
                continue
            results.solution = solution
            results.eigenvalues = getattr(solution, "evs", None)
        elif name in {"simul", "simulate"}:
            mode = options.get("mode", "random" if name == "simul" else "auto")
            if model.is_deterministic or mode in {"transition", "deterministic"}:
                from .solver import deterministic_solve

                det_opts = {k: v for k, v in options.items() if k != "mode"}
                sim = deterministic_solve(model, **det_opts)
                results.simulation = sim
                if hasattr(sim, "attrs") and not sim.attrs.get("converged", True):
                    results.add_warning(
                        f"Deterministic / perfect foresight simulation did not converge after "
                        f"{sim.attrs.get('iterations', '?')} iterations "
                        f"(maximum residual: {sim.attrs.get('residual', float('nan')):.2e}). "
                        "The computed solution is incorrect."
                    )
            else:
                solve_options = {
                    k: v for k, v in options.items() if k in {"method"}
                }
                solution = results.solution
                if solution is None or not hasattr(solution, "X"):
                    solution = _solve_or_record(solve_options)
                    if solution is None:
                        continue
                    results.solution = solution
                results.eigenvalues = getattr(solution, "evs", None)
                horizon = int(
                    options.get("T", options.get("irf", options.get("periods", 40)))
                )
                units = options.get("units", options.get("type", "deviation"))
                n_draws = int(options.get("N", 1))
                sim_options = {
                    k: v
                    for k, v in options.items()
                    if k in {"shocks", "initial_states", "rng"}
                }
                results.simulation = solution.simulate(
                    T=horizon,
                    mode=mode,
                    N=n_draws,
                    units=units,
                    **sim_options,
                )
        elif name in {"analyze", "stoch_simul"}:
            from .plots import plot_simulation

            plot_variables = options.get("variables", options.get("vars"))
            plot_options: dict[str, Any] = {"engine": "altair"}
            if plot_variables:
                plot_options["variables"] = plot_variables
            if name == "stoch_simul":
                # Dynare semantics: ``periods`` is the length of the
                # stochastic simulation, not the IRF horizon.
                horizon = int(options.get("T", options.get("irf", 40)))
                show_graph = not options.get("nograph", False) and horizon > 0
            else:
                horizon = int(
                    options.get("T", options.get("irf", options.get("periods", 40)))
                )
                show_graph = True

            if model.is_deterministic:
                from .solver import deterministic_solve

                det_opts = {
                    k: v
                    for k, v in options.items()
                    if k
                    not in {
                        "irf",
                        "periods",
                        "type",
                        "variables",
                        "vars",
                        "nograph",
                    }
                }
                det_opts["T"] = horizon
                sim = deterministic_solve(model, **det_opts)
                results.simulation = sim
                if hasattr(sim, "attrs") and not sim.attrs.get("converged", True):
                    results.add_warning(
                        f"Deterministic / perfect foresight simulation did not converge after "
                        f"{sim.attrs.get('iterations', '?')} iterations "
                        f"(maximum residual: {sim.attrs.get('residual', float('nan')):.2e}). "
                        "The computed solution is incorrect."
                    )
            else:
                solve_options = {
                    k: v for k, v in options.items() if k in {"method"}
                }
                solution = results.solution
                if solution is None or not hasattr(solution, "X"):
                    solution = _solve_or_record(solve_options)
                    if solution is None:
                        continue
                    results.solution = solution
                results.eigenvalues = getattr(solution, "evs", None)
                irf_type = options.get("units", options.get("type", "deviation"))
                results.moments = solution.moments()[1]
                if horizon > 0:
                    results.simulation = solution.irfs(type=irf_type, T=horizon)
            if show_graph and results.simulation is not None:
                results._plot_options = plot_options
                results.figure = plot_simulation(
                    results.simulation, engine="altair", variables=plot_variables
                )
        elif name == "plot":
            if results.simulation is not None:
                from .plots import plot_simulation

                plot_options = dict(options)
                if "vars" in plot_options and "variables" not in plot_options:
                    plot_options["variables"] = plot_options.pop("vars")
                if "type" in plot_options and "units" not in plot_options:
                    plot_options["units"] = plot_options.pop("type")
                engine = plot_options.get("engine", "altair")
                plot_opts = {k: v for k, v in plot_options.items() if k != "engine"}
                results._plot_options = dict(plot_options)  # type: ignore[attr-defined]
                results.figure = plot_simulation(
                    results.simulation, engine=engine, **plot_opts
                )
        else:
            raise NotImplementedError(f"Unsupported DynoModel.run command: {name}")

    # Add line-level warnings for non-zero residuals
    r = results.residuals
    if r is not None and r.size > 0 and abs(r).max() >= 1e-6:
        inds = np.where(abs(r) >= 1e-6)[0]
        for i in inds:
            tree = model.symbolic.equations[i]
            results.add_warning(str(r[i]), line=tree.meta.line)

    results.finish()
    return results

Bases: AbstractModel

Source code in src/dyno/dynare/model.py
 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
class DynareModel(AbstractModel):

    symbolic: Any
    _check_eigenvalues: bool = True

    @property
    def metadata(self: Self) -> dict[str, Any]:
        """Metadata is stored on `context` since the compiled `symbolic` object cannot hold arbitrary attributes."""
        context = getattr(self, "context", None)
        if context is None:
            return {}
        return context.get("metadata", {})

    @metadata.setter
    def metadata(self: Self, value: dict[str, Any]) -> None:
        context = getattr(self, "context", None)
        if context is None:
            raise AttributeError("Cannot set metadata before context is initialized")
        context["metadata"] = value

    @staticmethod
    def _load_dynare_preprocessor():
        try:
            from dynare_preprocessor import DynareModel as Modfile
            from dynare_preprocessor import PreprocessorException
        except ModuleNotFoundError as e:
            if e.name == "dynare_preprocessor":
                raise ModuleNotFoundError(
                    "DynareModel emulation requires the optional dependency "
                    "'dynare-preprocessor-pylib'. With pixi, add it via: "
                    "pixi add --feature dynare dynare-preprocessor-pylib, "
                    "or use an environment that includes the dynare feature. "
                    "Note: .mod files can still be imported with DynoModel without this dependency."
                ) from e
            raise
        return Modfile, PreprocessorException

    def _equation_line_numbers(self: Self) -> list[int]:
        """Best-effort mapping from equation index to source line number."""
        txt = getattr(self, "_original_txt", None)
        if not isinstance(txt, str):
            return []

        lines = txt.splitlines()
        in_model = False
        eq_lines: list[int] = []

        for idx, raw in enumerate(lines, start=1):
            stripped = raw.strip()
            if not in_model:
                # Matches "model;" and "model(linear);" forms.
                if stripped.lower().startswith("model"):
                    in_model = True
                continue

            if stripped.lower() == "end;":
                break

            if not stripped:
                continue
            if stripped.startswith(("//", "%", "#")):
                continue

            # Keep only likely equation lines.
            if "=" in stripped:
                eq_lines.append(idx)

        return eq_lines

    def _rebuild(self: Self) -> Self:
        txt = getattr(self, "_original_txt", None)
        if txt is None:
            with open(self.filename, "rt", encoding="utf-8") as f:
                txt = f.read()

        options = getattr(self, "_import_options", {})
        model = self.__class__(
            filename=self.filename,
            txt=txt,
            strict=getattr(self, "strict", False),
            **options,
        )

        previous = getattr(self, "_calibration_overrides", {})
        if len(previous) > 0:
            model = model.recalibrate(**previous)

        return model

    def recalibrate(self: Self, **calib):
        m = self._rebuild()

        known = set(m.context["constants"].keys()) | set(
            m.context["steady_states"].keys()
        )
        unknown = [k for k in calib.keys() if k not in known]
        if len(unknown) > 0:
            raise KeyError(f"Unknown calibration key(s): {', '.join(unknown)}")

        for key, value in calib.items():
            val = float(value)
            if key in m.context["constants"]:
                m.context["constants"][key] = val
            if key in m.context["steady_states"]:
                m.context["steady_states"][key] = val

        m.__steady_state__ = m.context["steady_states"]

        previous = getattr(self, "_calibration_overrides", {})
        m._calibration_overrides = previous | {k: float(v) for k, v in calib.items()}
        return m

    def run(self: Self, default_pipeline: bool = False) -> "RunResults":
        """Execute the model's computing commands and collect the results.

        Commands are read from ``model.metadata["dynare_commands"]`` (the
        computing commands of the ``.mod`` file) or, failing that, from
        ``model.metadata["run"]``, as normalized
        ``{"command": ..., "options": {...}}`` entries in the same stable
        ``@run:`` command format as :meth:`dyno.DynoModel.run`, and executed
        in order.

        Supported commands:

        - ``steady``: solve for the steady state (``model.steady(**options)``)
        - ``resid``: compute steady-state residuals
        - ``check``: compute residuals and check Blanchard-Kahn conditions
        - ``simul`` / ``simulate``: perfect-foresight simulation
        - ``stoch_simul``: solve, compute moments and IRFs (``irf``,
          ``type``, ``variables`` and ``nograph`` options)
        - ``plot``: select the variables to plot

        Other commands are ignored.

        Parameters
        ----------
        default_pipeline : bool, optional
            If True and the model defines no commands, compute residuals,
            then IRFs (stochastic models) or a perfect-foresight simulation
            (deterministic models) over 40 periods. By default False.

        Returns
        -------
        RunResults
            Container with the final model, residuals, solution, eigenvalues,
            moments, simulation and figure produced by the pipeline.
        """
        from dyno.report import RunResults

        commands = self.metadata.get("dynare_commands", self.metadata.get("run", []))
        model = self
        results = RunResults(model=model)
        results._from_pipeline = True
        plot_variables: list[str] | None = None
        show_graph = True

        if not commands and default_pipeline:
            # Default pipeline: residuals + solve + IRFs
            results.residuals = model.residuals
            if not model.is_deterministic:
                dr = model.perturb()
                results.solution = dr
                results.eigenvalues = dr.evs
                results.moments = dr.moments()[1]
                results.simulation = dr.irfs(type="deviation", T=40)
            else:
                from dyno.solver import deterministic_solve

                sim = deterministic_solve(model, T=40)
                results.simulation = {"Perfect Foresight": sim}
                if hasattr(sim, "attrs") and not sim.attrs.get("converged", True):
                    results.add_warning(
                        f"Deterministic / perfect foresight simulation did not converge after "
                        f"{sim.attrs.get('iterations', '?')} iterations "
                        f"(maximum residual: {sim.attrs.get('residual', float('nan')):.2e}). "
                        "The computed solution is incorrect."
                    )
        else:
            for cmd in commands:
                name = str(cmd.get("command", "")).lower()
                options = cmd.get("options", {})
                try:
                    if name == "steady":
                        model = model.steady(**options)
                        results.model = model
                        results.steady_stats = getattr(model, "steady_stats", None)
                    elif name == "check":
                        results.residuals = model.residuals
                        model = model.check()
                        results.eigenvalues = getattr(model, "_eigenvalues", None)
                    elif name == "resid":
                        results.residuals = model.residuals
                    elif name in {"simul", "simulate"}:
                        from dyno.solver import deterministic_solve

                        det_opts = {k: v for k, v in options.items() if k != "mode"}
                        sim = deterministic_solve(model, **det_opts)
                        results.simulation = sim
                        if not sim.attrs.get("converged", True):
                            results.add_warning(
                                f"Deterministic / perfect foresight simulation did not converge after "
                                f"{sim.attrs.get('iterations', '?')} iterations "
                                f"(maximum residual: {sim.attrs.get('residual', float('nan')):.2e}). "
                                "The computed solution is incorrect."
                            )
                    elif name == "plot":
                        plot_variables = options.get("variables") or None
                    elif name == "stoch_simul":
                        dr = model.perturb()
                        results.solution = dr
                        results.eigenvalues = dr.evs
                        results.moments = dr.moments()[1]
                        irf_type = options.get("type", "deviation")
                        # ``periods`` is the simulation length in Dynare, not
                        # the IRF horizon.
                        horizon = int(options.get("irf", 40))
                        plot_variables = options.get("variables")
                        show_graph = not options.get("nograph", False) and horizon > 0
                        if horizon > 0:
                            results.simulation = dr.irfs(type=irf_type, T=horizon)
                except SteadyStateError as e:
                    results.residuals = e.residuals
                    results.steady_stats = getattr(e, "steady_stats", None)
                    results.add_warning(str(e))
                    break

        # Add line-level warnings for non-zero residuals where we can map lines.
        r = results.residuals
        if r is not None and r.size > 0 and abs(r).max() >= 1e-6:
            inds = np.where(abs(r) >= 1e-6)[0]
            eq_lines = model._equation_line_numbers()
            for i in inds:
                line = eq_lines[i] if i < len(eq_lines) else None
                results.add_warning(
                    f"Equation {i + 1}: residual = {float(r[i]):.3e}",
                    line=line,
                )

        if show_graph and results.simulation is not None:
            from dyno.plots import plot_simulation

            results._plot_options = {"engine": "altair"}
            if plot_variables:
                results._plot_options["variables"] = plot_variables
            results.figure = plot_simulation(
                results.simulation, engine="altair", variables=plot_variables
            )

        results.finish()
        return results

    def import_model(
        self: Self,
        txt: str,
        **kwargs: Any,
    ) -> None:
        """imports model written in `.mod` format into symbolic attribute using Dynare's preprocessor

        Parameters
        ----------
        txt : str
            the model being imported in `.mod` form
        deriv_order : int, optional
            derivative order, by default 1
        params_deriv_order : int, optional
            parameters derivative order, by default 0
        allow_undeclared_params : bool, optional
            if True, automatically declare parameters that are assigned values
            without being explicitly declared in the parameters section, by default False
        """
        Modfile, PreprocessorException = self._load_dynare_preprocessor()

        deriv_order: int = kwargs.get("deriv_order", 1)
        params_deriv_order: int = kwargs.get("params_deriv_order", 0)
        if "allow_undeclared_params" in kwargs:
            allow_undeclared_params: bool = kwargs["allow_undeclared_params"]
        else:
            allow_undeclared_params = not getattr(self, "strict", False)

        self._import_options = {
            "deriv_order": deriv_order,
            "params_deriv_order": params_deriv_order,
            "allow_undeclared_params": allow_undeclared_params,
        }

        # Keep original modfile text for rebuilding immutable model variants.
        self._original_txt = txt

        # Preprocess to declare undeclared parameters if needed
        if allow_undeclared_params:
            txt = self._declare_undeclared_params(txt)

        # The binding runs the preprocessor in a stochastic context, which
        # Dynare refuses to mix with ``perfect_foresight_solver``/``simul``.
        # Those statements are removed here and re-inserted as run commands.
        txt, self._perfect_foresight_statements = _extract_solver_statements(txt)

        try:
            self.symbolic = Modfile(txt, deriv_order, params_deriv_order)
        except PreprocessorException as e:
            raise DynareParserError(e) from e

    def _extract_dynare_commands(self: Self) -> list[dict[str, Any]]:
        """Extract Dynare statements/commands from the preprocessor JSON.

        Returns
        -------
        list[dict[str, Any]]
            Ordered list of command dictionaries with ``command`` and ``options``.
        """
        import json

        payload = json.loads(self.symbolic.json_string)
        transformed = payload.get("transformed_modfile", {})
        statements = transformed.get("statements", [])

        ignored = {"param_init", "initval"}

        commands: list[dict[str, Any]] = []
        for statement in statements:
            command = statement.get("statementName")
            if not isinstance(command, str):
                continue
            if command in ignored:
                continue

            options = statement.get("options", {})
            if not isinstance(options, dict):
                options = {}
            options = {k: _coerce_option(v) for k, v in options.items()}

            symbol_list = statement.get("symbol_list")
            if isinstance(symbol_list, list) and symbol_list:
                options = {**options, "variables": [str(s) for s in symbol_list]}

            commands.append({"command": command, "options": options})

        solver_statements = list(getattr(self, "_perfect_foresight_statements", []))
        if solver_statements:
            setups = [
                i
                for i, c in enumerate(commands)
                if c["command"] == "perfect_foresight_setup"
            ]
            at = setups[-1] + 1 if setups else len(commands)
            commands[at:at] = solver_statements

        from dyno.larkfiles import _translate_perfect_foresight_commands

        return _translate_perfect_foresight_commands(commands)

    def _deterministic_shocks(self: Self) -> dict[str, dict[int, float]]:
        """Forced exogenous paths from the ``shocks`` blocks of the preprocessor JSON.

        ``var x; periods p1:p2; values v;`` becomes ``{x: {t: v}}`` for
        ``t`` in ``p1..p2``; date 0 is the initial condition, as in .dyno files.
        """
        import json

        payload = json.loads(self.symbolic.json_string)
        statements = payload.get("transformed_modfile", {}).get("statements", [])
        constants = {
            k: v
            for k, v in self.symbolic.context.items()
            if k in self.symbolic.parameters
        }

        values: dict[str, dict[int, float]] = {}
        for statement in statements:
            if statement.get("statementName") != "shocks":
                continue
            for entry in statement.get("deterministic_shocks", []):
                series = values.setdefault(str(entry["var"]), {})
                for item in entry.get("values", []):
                    value = _evaluate_expression(str(item["value"]), constants)
                    for t in range(int(item["period1"]), int(item["period2"]) + 1):
                        series[t] = value
        return values

    def _declare_undeclared_params(self: Self, txt: str) -> str:
        """Automatically declare parameters that are assigned values without being declared.

        Parameters
        ----------
        txt : str
            the model text in `.mod` format

        Returns
        -------
        str
            the modified model text with undeclared parameters declared
        """
        import re

        # Remove comments from text for parsing
        txt_no_comments = re.sub(r"/\*.*?\*/", "", txt, flags=re.DOTALL)
        txt_no_comments = re.sub(r"//.*?$", "", txt_no_comments, flags=re.MULTILINE)

        # Only look at declarations before the model block
        # Split by 'model;' and take only the declarations part
        model_split = re.split(r"\bmodel\s*;", txt_no_comments, flags=re.IGNORECASE)
        declarations_part = model_split[0] if model_split else txt_no_comments

        # Remove shocks block from declarations part if placed before model
        declarations_part = re.sub(
            r"\bshocks\s*;.*?end\s*;",
            "",
            declarations_part,
            flags=re.DOTALL | re.IGNORECASE,
        )

        # Split original text into lines for reconstruction
        lines = txt.split("\n")

        # Find all declared identifiers (vars, varexo, parameters)
        declared_identifiers = set()
        params_section_idx = None

        # Extract var declarations (only before model block)
        var_matches = re.findall(
            r"^\s*var\s+(.*?);\s*$", declarations_part, re.MULTILINE | re.IGNORECASE
        )
        for match in var_matches:
            declared_identifiers.update(re.findall(r"\b([a-zA-Z_]\w*)\b", match))

        # Extract varexo declarations (only before model block)
        varexo_matches = re.findall(
            r"^\s*varexo\s+(.*?);\s*$", declarations_part, re.MULTILINE | re.IGNORECASE
        )
        for match in varexo_matches:
            declared_identifiers.update(re.findall(r"\b([a-zA-Z_]\w*)\b", match))

        # Extract parameters declarations and find section (only before model block)
        params_matches = re.findall(
            r"^\s*parameters\s+(.*?);\s*$",
            declarations_part,
            re.MULTILINE | re.IGNORECASE,
        )
        for match in params_matches:
            declared_identifiers.update(re.findall(r"\b([a-zA-Z_]\w*)\b", match))

        # Find parameters section index in original file
        for i, line in enumerate(lines):
            if re.match(r"\s*parameters\s", line, re.IGNORECASE):
                params_section_idx = i
                break

        # Find parameter assignments (identifier = number;) that are not yet declared
        # Only look at assignments before model block
        undeclared = set()
        assignment_pattern = r"^\s*([a-zA-Z_]\w*)\s*=\s*[+-]?[\d.eE+-]+(?:\s*[;]|$)"

        for line in declarations_part.split("\n"):
            match = re.match(assignment_pattern, line)
            if match:
                param_name = match.group(1)
                if param_name not in declared_identifiers:
                    undeclared.add(param_name)

        # If there are undeclared parameters, add them to the declaration
        if undeclared:
            undeclared_list = ", ".join(sorted(undeclared))

            if params_section_idx is not None:
                # There's already a parameters section - append to it
                # Find the semicolon at the end of the parameters declaration
                j = params_section_idx
                while j < len(lines):
                    if ";" in lines[j]:
                        # Insert before the semicolon
                        lines[j] = lines[j].replace(";", f", {undeclared_list};", 1)
                        break
                    j += 1
            else:
                # No parameters section exists, create one after var/varexo declarations
                # Find the last var/varexo declaration (ignoring any inside shocks blocks)
                last_var_section = 0
                in_shocks = False
                for i, line in enumerate(lines):
                    stripped = line.strip().lower()
                    if stripped.startswith("shocks;"):
                        in_shocks = True
                    elif in_shocks and stripped.startswith("end;"):
                        in_shocks = False
                        continue
                    if not in_shocks and re.match(
                        r"\s*(var|varexo)\s", line, re.IGNORECASE
                    ):
                        last_var_section = i
                        # Find end of this declaration
                        j = i
                        while j < len(lines) and ";" not in lines[j]:
                            j += 1
                        if j < len(lines):
                            last_var_section = j

                # Insert parameters section after the last var/varexo
                insert_idx = last_var_section + 1
                lines.insert(insert_idx, f"\nparameters {undeclared_list};")

        return "\n".join(lines)

    def _set_context(self: Self) -> None:
        """retrieves calibration values"""

        c = self.symbolic.context  # dynare preprocessor context
        endogenous = self.symbolic.endogenous
        exogenous = self.symbolic.exogenous
        variables = endogenous + exogenous
        parameters = self.symbolic.parameters

        steady_states = {
            k: v for (k, v) in c.items() if (k in endogenous) or (k in exogenous)
        }
        constants = {k: v for (k, v) in c.items() if (k in parameters)}

        commands = self._extract_dynare_commands()
        deterministic_shocks = self._deterministic_shocks()

        # A file is a perfect-foresight model when it forces exogenous paths
        # or uses the perfect-foresight commands, and declares no shock
        # variances nor calls stoch_simul (Dynare refuses to mix the two).
        names = {c["command"] for c in commands}
        isdeterministic = (bool(deterministic_shocks) or "simul" in names) and not (
            len(self.symbolic.covariances) > 0 or "stoch_simul" in names
        )
        exo = exogenous

        values: dict[str, dict[int, float]]
        if isdeterministic:
            values = deterministic_shocks
            processes = {}
        else:
            n = len(exo)
            covar = np.zeros((n, n))
            index = {name: i for (i, name) in enumerate(exo)}
            for (var1, var2), val in self.symbolic.covariances.items():
                covar[index[var1], index[var2]] = val
                covar[index[var2], index[var1]] = val
            # self.processes =
            values = {}
            processes = {tuple(exo): Normal(Σ=covar)}

        context = {
            "constants": constants,
            "variables": {v: {} for v in variables},
            "values": values,
            "processes": processes,
            "steady_states": steady_states,
            "metadata": {
                "dynare_commands": commands,
            },
        }
        # self.paths = None
        # self.exogenous = self.processes
        self.context = cast(ModelContext, context)

    @property
    def equations(self):

        return self.symbolic.equations

    def compute_residuals(self, y1, y2, y3, e):
        p = [self.context["constants"][p] for p in self.symbolic.parameters]
        y, e = self.__steady_state_vectors__
        return self._f_dynamic(y, y, y, e, p)

    def compute_jacobians(self, y1, y2, y3, e):
        p = [self.context["constants"][p] for p in self.symbolic.parameters]
        y, e = self.__steady_state_vectors__
        return self._f_dynamic(y, y, y, e, p, diff=True)

    def compute_derivatives(self: Self):

        y = [self.steady_state[v] for v in self.symbols["endogenous"]]
        e = [self.steady_state[v] for v in self.symbols["exogenous"]]
        p = [self.context["constants"][v] for v in self.symbols["parameters"]]

        return self.symbolic.derivatives(y, y, y, e, e, p)

    def _dynamic_point(self, v_prev, v_curr, v_next, diff=False):
        q = len(self.symbols["endogenous"])
        p = [self.context["constants"][name] for name in self.symbolic.parameters]
        e = v_curr[q:]
        if not diff:
            return np.asarray(self._f_dynamic(v_next[:q], v_curr[:q], v_prev[:q], e, p))
        r, A, B, C, D = self._f_dynamic(
            v_next[:q], v_curr[:q], v_prev[:q], e, p, diff=True
        )
        n_vars = len(self.symbols["variables"])
        J = np.zeros((len(r), n_vars, 3))
        J[:, :q, 0] = C
        J[:, :q, 1] = B
        J[:, :q, 2] = A
        J[:, q:, 1] = D
        return np.asarray(r), J

    def _f_dynamic(
        self: Self,
        y0: TVector,
        y1: TVector,
        y2: TVector,
        e: TVector,
        p: TVector,
        diff: bool = False,
    ) -> TVector | tuple[TVector, TMatrix, TMatrix, TMatrix, TMatrix]:
        """function f describing the behavior of the dynamic system $f(y_{t+1}, y_t, y_{t-1}, ε_t, p) = 0$

        Parameters
        ----------
        y0,y1,y2 : Vector
            the system's endogenous variable values at times t+1, t and t-1 respectively
        e : Vector
            exogenous variable values
        p : Vector
            parameter values
        diff : bool, optional
            if set to True returns the function's partial derivatives with regards to y0, y1, y2 and e as well, by default False

        Returns
        -------
        Vector|tuple[Vector, Matrix, Matrix, Matrix, Matrix]
            value of f(y0, y1, y2, e, p), as well as partial derivatives w.r.t. y0, y1, y2 and e if diff is set to True
        """

        # (endo_future, endo_present, endo_past, exo, exo_det, params).
        # ``varexo_det`` variables are not supported, so that slot is empty.
        args: list[list[Any]] = [
            list(y0),
            list(y1),
            list(y2),
            list(e),
            [],
            list(p),
        ]

        r = np.array(self.symbolic.residuals(*args))

        if diff:
            jacobians = self.symbolic.jacobians(*args)
            del jacobians[4]
            n = len(self.equations)
            lengths = [n] * 3 + [len(e), len(p)]
            r1, r2, r3, r4 = [
                sparse_to_dense(n, length, j)
                for (j, length) in zip(jacobians[:-1], lengths)
            ]
            return r, r1, r2, r3, r4

        return r

metadata property writable

Metadata is stored on context since the compiled symbolic object cannot hold arbitrary attributes.

import_model(txt, **kwargs)

imports model written in .mod format into symbolic attribute using Dynare's preprocessor

Parameters:

Name Type Description Default
txt str

the model being imported in .mod form

required
deriv_order int

derivative order, by default 1

required
params_deriv_order int

parameters derivative order, by default 0

required
allow_undeclared_params bool

if True, automatically declare parameters that are assigned values without being explicitly declared in the parameters section, by default False

required
Source code in src/dyno/dynare/model.py
def import_model(
    self: Self,
    txt: str,
    **kwargs: Any,
) -> None:
    """imports model written in `.mod` format into symbolic attribute using Dynare's preprocessor

    Parameters
    ----------
    txt : str
        the model being imported in `.mod` form
    deriv_order : int, optional
        derivative order, by default 1
    params_deriv_order : int, optional
        parameters derivative order, by default 0
    allow_undeclared_params : bool, optional
        if True, automatically declare parameters that are assigned values
        without being explicitly declared in the parameters section, by default False
    """
    Modfile, PreprocessorException = self._load_dynare_preprocessor()

    deriv_order: int = kwargs.get("deriv_order", 1)
    params_deriv_order: int = kwargs.get("params_deriv_order", 0)
    if "allow_undeclared_params" in kwargs:
        allow_undeclared_params: bool = kwargs["allow_undeclared_params"]
    else:
        allow_undeclared_params = not getattr(self, "strict", False)

    self._import_options = {
        "deriv_order": deriv_order,
        "params_deriv_order": params_deriv_order,
        "allow_undeclared_params": allow_undeclared_params,
    }

    # Keep original modfile text for rebuilding immutable model variants.
    self._original_txt = txt

    # Preprocess to declare undeclared parameters if needed
    if allow_undeclared_params:
        txt = self._declare_undeclared_params(txt)

    # The binding runs the preprocessor in a stochastic context, which
    # Dynare refuses to mix with ``perfect_foresight_solver``/``simul``.
    # Those statements are removed here and re-inserted as run commands.
    txt, self._perfect_foresight_statements = _extract_solver_statements(txt)

    try:
        self.symbolic = Modfile(txt, deriv_order, params_deriv_order)
    except PreprocessorException as e:
        raise DynareParserError(e) from e

run(default_pipeline=False)

Execute the model's computing commands and collect the results.

Commands are read from model.metadata["dynare_commands"] (the computing commands of the .mod file) or, failing that, from model.metadata["run"], as normalized {"command": ..., "options": {...}} entries in the same stable @run: command format as :meth:dyno.DynoModel.run, and executed in order.

Supported commands:

  • steady: solve for the steady state (model.steady(**options))
  • resid: compute steady-state residuals
  • check: compute residuals and check Blanchard-Kahn conditions
  • simul / simulate: perfect-foresight simulation
  • stoch_simul: solve, compute moments and IRFs (irf, type, variables and nograph options)
  • plot: select the variables to plot

Other commands are ignored.

Parameters:

Name Type Description Default
default_pipeline bool

If True and the model defines no commands, compute residuals, then IRFs (stochastic models) or a perfect-foresight simulation (deterministic models) over 40 periods. By default False.

False

Returns:

Type Description
RunResults

Container with the final model, residuals, solution, eigenvalues, moments, simulation and figure produced by the pipeline.

Source code in src/dyno/dynare/model.py
def run(self: Self, default_pipeline: bool = False) -> "RunResults":
    """Execute the model's computing commands and collect the results.

    Commands are read from ``model.metadata["dynare_commands"]`` (the
    computing commands of the ``.mod`` file) or, failing that, from
    ``model.metadata["run"]``, as normalized
    ``{"command": ..., "options": {...}}`` entries in the same stable
    ``@run:`` command format as :meth:`dyno.DynoModel.run`, and executed
    in order.

    Supported commands:

    - ``steady``: solve for the steady state (``model.steady(**options)``)
    - ``resid``: compute steady-state residuals
    - ``check``: compute residuals and check Blanchard-Kahn conditions
    - ``simul`` / ``simulate``: perfect-foresight simulation
    - ``stoch_simul``: solve, compute moments and IRFs (``irf``,
      ``type``, ``variables`` and ``nograph`` options)
    - ``plot``: select the variables to plot

    Other commands are ignored.

    Parameters
    ----------
    default_pipeline : bool, optional
        If True and the model defines no commands, compute residuals,
        then IRFs (stochastic models) or a perfect-foresight simulation
        (deterministic models) over 40 periods. By default False.

    Returns
    -------
    RunResults
        Container with the final model, residuals, solution, eigenvalues,
        moments, simulation and figure produced by the pipeline.
    """
    from dyno.report import RunResults

    commands = self.metadata.get("dynare_commands", self.metadata.get("run", []))
    model = self
    results = RunResults(model=model)
    results._from_pipeline = True
    plot_variables: list[str] | None = None
    show_graph = True

    if not commands and default_pipeline:
        # Default pipeline: residuals + solve + IRFs
        results.residuals = model.residuals
        if not model.is_deterministic:
            dr = model.perturb()
            results.solution = dr
            results.eigenvalues = dr.evs
            results.moments = dr.moments()[1]
            results.simulation = dr.irfs(type="deviation", T=40)
        else:
            from dyno.solver import deterministic_solve

            sim = deterministic_solve(model, T=40)
            results.simulation = {"Perfect Foresight": sim}
            if hasattr(sim, "attrs") and not sim.attrs.get("converged", True):
                results.add_warning(
                    f"Deterministic / perfect foresight simulation did not converge after "
                    f"{sim.attrs.get('iterations', '?')} iterations "
                    f"(maximum residual: {sim.attrs.get('residual', float('nan')):.2e}). "
                    "The computed solution is incorrect."
                )
    else:
        for cmd in commands:
            name = str(cmd.get("command", "")).lower()
            options = cmd.get("options", {})
            try:
                if name == "steady":
                    model = model.steady(**options)
                    results.model = model
                    results.steady_stats = getattr(model, "steady_stats", None)
                elif name == "check":
                    results.residuals = model.residuals
                    model = model.check()
                    results.eigenvalues = getattr(model, "_eigenvalues", None)
                elif name == "resid":
                    results.residuals = model.residuals
                elif name in {"simul", "simulate"}:
                    from dyno.solver import deterministic_solve

                    det_opts = {k: v for k, v in options.items() if k != "mode"}
                    sim = deterministic_solve(model, **det_opts)
                    results.simulation = sim
                    if not sim.attrs.get("converged", True):
                        results.add_warning(
                            f"Deterministic / perfect foresight simulation did not converge after "
                            f"{sim.attrs.get('iterations', '?')} iterations "
                            f"(maximum residual: {sim.attrs.get('residual', float('nan')):.2e}). "
                            "The computed solution is incorrect."
                        )
                elif name == "plot":
                    plot_variables = options.get("variables") or None
                elif name == "stoch_simul":
                    dr = model.perturb()
                    results.solution = dr
                    results.eigenvalues = dr.evs
                    results.moments = dr.moments()[1]
                    irf_type = options.get("type", "deviation")
                    # ``periods`` is the simulation length in Dynare, not
                    # the IRF horizon.
                    horizon = int(options.get("irf", 40))
                    plot_variables = options.get("variables")
                    show_graph = not options.get("nograph", False) and horizon > 0
                    if horizon > 0:
                        results.simulation = dr.irfs(type=irf_type, T=horizon)
            except SteadyStateError as e:
                results.residuals = e.residuals
                results.steady_stats = getattr(e, "steady_stats", None)
                results.add_warning(str(e))
                break

    # Add line-level warnings for non-zero residuals where we can map lines.
    r = results.residuals
    if r is not None and r.size > 0 and abs(r).max() >= 1e-6:
        inds = np.where(abs(r) >= 1e-6)[0]
        eq_lines = model._equation_line_numbers()
        for i in inds:
            line = eq_lines[i] if i < len(eq_lines) else None
            results.add_warning(
                f"Equation {i + 1}: residual = {float(r[i]):.3e}",
                line=line,
            )

    if show_graph and results.simulation is not None:
        from dyno.plots import plot_simulation

        results._plot_options = {"engine": "altair"}
        if plot_variables:
            results._plot_options["variables"] = plot_variables
        results.figure = plot_simulation(
            results.simulation, engine="altair", variables=plot_variables
        )

    results.finish()
    return results