extendedmosaicperm package

Subpackages

• extendedmosaicperm.experiments package
  • Submodules
    • extendedmosaicperm.experiments.adaptive_tiling module
      • AdaptiveTilingExperiment
    • extendedmosaicperm.experiments.base module
      • BaseExperiment
    • extendedmosaicperm.experiments.plotting module
      • compute_power_table()
      • enrich_flatten()
      • parse_label()
      • plot_figure3_panels()
      • plot_qq_grid_all_sizes()
      • plot_qq_grid_generators_by_alpha()
    • extendedmosaicperm.experiments.ridge module
      • RidgeExperiment
    • extendedmosaicperm.experiments.sign_flip module
      • SignFlipExperiment
  • Module contents
    • AdaptiveTilingExperiment
      • AdaptiveTilingExperiment.extract_method()
      • AdaptiveTilingExperiment.run()
    • BaseExperiment
      • BaseExperiment.results
      • BaseExperiment.extract_method()
      • BaseExperiment.flatten()
      • BaseExperiment.load()
      • BaseExperiment.plot_power()
      • BaseExperiment.plot_pval_histogram()
      • BaseExperiment.plot_qq()
      • BaseExperiment.save()
      • BaseExperiment.summarize()
    • RidgeExperiment
      • RidgeExperiment.extract_method()
      • RidgeExperiment.run()
    • SignFlipExperiment
      • SignFlipExperiment.compare_method_timings_all_sizes()
      • SignFlipExperiment.extract_method()
      • SignFlipExperiment.run()
    • compute_power_table()
    • enrich_flatten()
    • plot_figure3_panels()
    • plot_qq_grid_all_sizes()
    • plot_qq_grid_generators_by_alpha()

Submodules

• extendedmosaicperm.factor module
  • ExtendMosaicFactorTest
    • ExtendMosaicFactorTest.sign_flipping
    • ExtendMosaicFactorTest.rng
    • ExtendMosaicFactorTest.ridge_alphas
    • ExtendMosaicFactorTest.compute_mosaic_residuals()
    • ExtendMosaicFactorTest.flip_sign_residuals()
• extendedmosaicperm.factor_data module
  • FactorModelDataGenerator
    • FactorModelDataGenerator.generate_data_block_correlation()
    • FactorModelDataGenerator.generate_data_diagonal_plus_common_factor()
    • FactorModelDataGenerator.generate_data_random_correlation()
• extendedmosaicperm.tilings module
  • build_adaptive_tiling()
  • estimate_covariance()
  • greedy_grouping()

Module contents

extendedmosaicperm: extensions and experiments for mosaic permutation tests.

class ExtendMosaicFactorTest(*args, sign_flipping=False, seed=123, ridge_alphas=None, adaptive_tiling=False, **kwargs)

Bases: MosaicFactorTest

Extended mosaic factor test.

Extends mosaicperm.factor.MosaicFactorTest by adding:

• sign-flip inference instead of standard permutation,

• RidgeCV-based residual estimation,

• adaptive tiling of data based on the correlation structure.

Parameters:

• *args – Positional arguments forwarded to the base mosaicperm.factor.MosaicFactorTest constructor.

• sign_flipping (bool) – Whether to use sign-flip inference instead of permutations.

• seed (int) – Seed for the internal random number generator used for sign flips.

• ridge_alphas (Optional[ndarray]) – Sequence of alpha values for sklearn.linear_model.RidgeCV. If None, ordinary least squares (OLS) is used to estimate residuals.

• adaptive_tiling (bool) – If True, tiles are built automatically from outcomes and exposures using extendedmosaicperm.tilings.build_adaptive_tiling().

• **kwargs – Keyword arguments forwarded to the base mosaicperm.factor.MosaicFactorTest constructor.

sign_flipping

Boolean flag indicating whether sign-flip inference is used.

rng

NumPy random number generator used for sign flips.

ridge_alphas

Grid of RidgeCV penalization strengths, or None for OLS.

Examples

Basic workflow with fixed exposures (2D exposures) and OLS residuals:

>>> import numpy as np
>>> from mosaicperm import statistics
>>> from extendedmosaicperm.factor import ExtendMosaicFactorTest
>>> from extendedmosaicperm.tilings import build_adaptive_tiling
>>>
>>> T, N, K = 60, 12, 3
>>> rng = np.random.default_rng(0)
>>> F = rng.standard_normal((T, K))
>>> B = rng.standard_normal((K, N))
>>> Y = F @ B + rng.standard_normal((T, N))
>>>
>>> tiles = build_adaptive_tiling(Y, B.T)  # exposures as (N x K)
>>> mpt = ExtendMosaicFactorTest(
...     outcomes=Y,
...     exposures=B.T,
...     tiles=tiles,
...     test_stat=statistics.mean_maxcorr_stat,
...     sign_flipping=True,     # enable sign-flip inference
...     ridge_alphas=None,      # OLS for residuals
...     seed=0,
... )
>>> _ = mpt.compute_mosaic_residuals()  
>>> p = mpt._compute_p_value(nrand=20, verbose=False)
>>> 0.0 <= p <= 1.0
True

Ridge-regularized residuals:

>>> import numpy as np
>>> from mosaicperm import statistics
>>> from extendedmosaicperm.factor import ExtendMosaicFactorTest
>>> from extendedmosaicperm.tilings import build_adaptive_tiling
>>>
>>> T, N, K = 60, 12, 3
>>> rng = np.random.default_rng(1)
>>> F = rng.standard_normal((T, K))
>>> B = rng.standard_normal((K, N))
>>> Y = F @ B + rng.standard_normal((T, N))
>>>
>>> tiles = build_adaptive_tiling(Y, B.T)
>>> alphas = np.logspace(-3, 3, 5)
>>> mpt_ridge = ExtendMosaicFactorTest(
...     outcomes=Y,
...     exposures=B.T,
...     tiles=tiles,
...     test_stat=statistics.mean_maxcorr_stat,
...     ridge_alphas=alphas,    # use RidgeCV
...     sign_flipping=False,    # standard permutation
...     seed=1,
... )
>>> _ = mpt_ridge.compute_mosaic_residuals()  
>>> p_ridge = mpt_ridge._compute_p_value(nrand=20, verbose=False)
>>> 0.0 <= p_ridge <= 1.0
True

Adaptive tiling (no need to pass tiles explicitly):

>>> import numpy as np
>>> from mosaicperm import statistics
>>> from extendedmosaicperm.factor import ExtendMosaicFactorTest
>>>
>>> T, N, K = 60, 12, 3
>>> rng = np.random.default_rng(2)
>>> F = rng.standard_normal((T, K))
>>> B = rng.standard_normal((K, N))
>>> Y = F @ B + rng.standard_normal((T, N))
>>>
>>> mpt_adapt = ExtendMosaicFactorTest(
...     outcomes=Y,
...     exposures=B.T,
...     adaptive_tiling=True,
...     test_stat=statistics.mean_maxcorr_stat,
...     sign_flipping=True,
...     seed=2,
... )
>>> _ = mpt_adapt.compute_mosaic_residuals()  
>>> p_adapt = mpt_adapt._compute_p_value(nrand=20, verbose=False)
>>> 0.0 <= p_adapt <= 1.0
True

compute_mosaic_residuals()

Compute mosaic residuals for all tiles.

For each (batch, group) tile, factor loadings are estimated and residuals are computed. If ridge_alphas is None, residuals are obtained via OLS; otherwise, a separate RidgeCV is fit for each observation in the tile.

Returns:

Residual matrix with the same shape as self.outcomes.

Return type:

np.ndarray

flip_sign_residuals()

Apply random sign flips within each tile.

For each tile (batch × group), an i.i.d. Rademacher vector of length len(batch) is drawn and applied across all assets in the tile.

Return type:

None

class FactorModelDataGenerator(n_timepoints=60, n_assets=10, n_factors=3, seed=123)

Bases: object

Generator of synthetic data for factor-model simulations.

Supports multiple residual correlation structures, time-varying exposures, and configurable distributional asymmetry.

Parameters:

• n_timepoints (int) – Number of time periods T.

• n_assets (int) – Number of assets N.

• n_factors (int) – Number of factors K.

• seed (int) – Random seed for reproducibility.

Examples

Basic usage with constant exposures and symmetric residuals:

>>> from extendedmosaicperm.factor_data import FactorModelDataGenerator
>>> gen = FactorModelDataGenerator(n_timepoints=20, n_assets=6, n_factors=2, seed=0)
>>> Y, L_time, X, eps = gen.generate_data_random_correlation(
...     violation_strength=0.1,
...     exposures_update_interval=None,
...     symmetric_residuals=True,
... )
>>> Y.shape, L_time.shape, X.shape, eps.shape
((20, 6), (20, 6, 2), (20, 2), (20, 6))

Time-varying exposures, redrawn every 5 periods:

>>> gen = FactorModelDataGenerator(n_timepoints=15, n_assets=5, n_factors=3, seed=1)
>>> Y, L_time, X, eps = gen.generate_data_block_correlation(
...     violation_strength=0.2,
...     block_ratio=0.4,
...     exposures_update_interval=5,
...     symmetric_residuals=True,
... )
>>> L_time[0].shape
(5, 3)

Asymmetric residuals (gamma-based, standardized):

>>> gen = FactorModelDataGenerator(n_timepoints=12, n_assets=7, n_factors=2, seed=2)
>>> Y, L_time, X, eps = gen.generate_data_diagonal_plus_common_factor(
...     violation_strength=0.3,
...     exposures_update_interval=None,
...     symmetric_residuals=False,
...     gamma_shape=2.0,
...     gamma_scale=1.0,
... )
>>> eps.shape == Y.shape == (12, 7)
True

generate_data_block_correlation(violation_strength=0.0, block_ratio=0.5, exposures_update_interval=None, symmetric_residuals=True, gamma_shape=2.0, gamma_scale=1.0)

Generate data with block-correlated residuals.

A subset of assets (a “block”) exhibits stronger cross-sectional correlation than the rest. The size of the block and correlation strength are controlled by block_ratio and violation_strength.

Parameters:

• violation_strength (float) – Within-block correlation strength in [0, 1].

• block_ratio (float) – Fraction of assets forming the strongly correlated block in (0, 1].

• exposures_update_interval (Optional[int]) – Interval (in timepoints) for refreshing exposures; None means constant exposures.

• symmetric_residuals (bool) – Whether residuals are symmetric (Gaussian) or asymmetric (gamma-based).

• gamma_shape (float) – Shape parameter for asymmetric (Gamma) noise.

• gamma_scale (float) – Scale parameter for asymmetric (Gamma) noise.

Returns:

A tuple (Y, L_time, X, eps) where:

• Y – simulated outcomes, shape (T, N),

• L_time – exposures, shape (T, N, K),

• X – latent factors, shape (T, K),

• eps – residuals, shape (T, N).

Return type:

tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]

Examples

>>> from extendedmosaicperm.factor_data import FactorModelDataGenerator
>>> gen = FactorModelDataGenerator(n_timepoints=24, n_assets=10, n_factors=3, seed=0)
>>> Y, L_time, X, eps = gen.generate_data_block_correlation(
...     violation_strength=0.3, block_ratio=0.5, exposures_update_interval=6
... )
>>> Y.shape[1], L_time.shape[2], X.shape[1]
(10, 3, 3)

generate_data_diagonal_plus_common_factor(violation_strength=0.0, exposures_update_interval=None, symmetric_residuals=True, gamma_shape=2.0, gamma_scale=1.0)

Generate data with diagonal residual covariance plus a common factor.

Residuals are decomposed into a diagonal idiosyncratic component and a single common factor component with strength controlled by violation_strength.

Parameters:

• violation_strength (float) – Strength of the common residual factor in [0, 1].

• exposures_update_interval (Optional[int]) – Interval (in timepoints) for refreshing exposures; None means constant exposures.

• symmetric_residuals (bool) – Whether residuals are symmetric (Gaussian) or asymmetric (gamma-based).

• gamma_shape (float) – Shape parameter for asymmetric (Gamma) noise.

• gamma_scale (float) – Scale parameter for asymmetric (Gamma) noise.

Returns:

A tuple (Y, L_time, X, eps) where:

• Y – simulated outcomes, shape (T, N),

• L_time – exposures, shape (T, N, K),

• X – latent factors, shape (T, K),

• eps – residuals, shape (T, N).

Return type:

tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]

Examples

>>> from extendedmosaicperm.factor_data import FactorModelDataGenerator
>>> gen = FactorModelDataGenerator(n_timepoints=25, n_assets=9, n_factors=2, seed=0)
>>> Y, L_time, X, eps = gen.generate_data_diagonal_plus_common_factor(
...     violation_strength=0.2
... )
>>> X.shape
(25, 2)

generate_data_random_correlation(violation_strength=0.0, exposures_update_interval=None, symmetric_residuals=True, gamma_shape=2.0, gamma_scale=1.0)

Generate data with a random residual correlation structure.

Residuals are generated from a random positive-definite correlation matrix, with the degree of cross-sectional dependence controlled by violation_strength.

Parameters:

• violation_strength (float) – Strength of cross-sectional residual correlation in [0, 1]. 0 corresponds to independence, 1 to the raw random correlation structure.

• exposures_update_interval (Optional[int]) – Interval (in timepoints) for refreshing exposures; None means constant exposures.

• symmetric_residuals (bool) – Whether residuals are symmetric (Gaussian) or asymmetric (gamma-based).

• gamma_shape (float) – Shape parameter for asymmetric (Gamma) noise.

• gamma_scale (float) – Scale parameter for asymmetric (Gamma) noise.

Returns:

A tuple (Y, L_time, X, eps) where:

• Y – simulated outcomes, shape (T, N),

• L_time – exposures, shape (T, N, K),

• X – latent factors, shape (T, K),

• eps – residuals, shape (T, N).

Return type:

tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]

Examples

>>> from extendedmosaicperm.factor_data import FactorModelDataGenerator
>>> gen = FactorModelDataGenerator(n_timepoints=30, n_assets=8, n_factors=3, seed=0)
>>> Y, L_time, X, eps = gen.generate_data_random_correlation(violation_strength=0.2)
>>> Y.shape, L_time.shape, X.shape, eps.shape
((30, 8), (30, 8, 3), (30, 3), (30, 8))

build_adaptive_tiling(outcomes, exposures, batch_size=10, D=None, seed=0)

Build an adaptive tiling of the sample based on residual covariance.

The time dimension is split into batches of size batch_size. For the first batch, groups are assigned using a random even partition. For subsequent batches, residuals from previous tiles are used to estimate a covariance matrix, and the grouping is updated via greedy_grouping().

Parameters:

• outcomes (ndarray) – Outcome matrix of shape (T, p).

• exposures (ndarray) – Factor loadings, either constant of shape (p, K) or time-varying of shape (T, p, K).

• batch_size (int) – Number of observations per batch.

• D (Optional[int]) – Number of groups. If None, defaults to max(2, p // (2 * K)), where K is the number of factors.

• seed (int) – Random seed used for initial grouping and subsequent updates.

Returns:

A tiling object containing all (batch, group) pairs.

Return type:

Tiling

Raises:

ValueError – If outcomes or exposures is None, or if the exposure slices cannot be reshaped to (p, K).

Examples

>>> import numpy as np
>>> from mosaicperm.tilings import Tiling
>>> from extendedmosaicperm.tilings import build_adaptive_tiling
>>> rng = np.random.default_rng(0)
>>> T, p, K = 12, 6, 2
>>> Y = rng.normal(size=(T, p))
>>> L = rng.normal(size=(p, K))
>>> til = build_adaptive_tiling(Y, L, batch_size=4, D=3, seed=7)
>>> isinstance(til, Tiling)
True
>>> len(til)  
9