Module fcmpy.intervention.interventionConstructor
Expand source code
import pandas as pd
import numpy as np
from abc import ABC, abstractmethod
class InterventionConstructor(ABC):
"""
Class of methods for building interventions.
"""
@abstractmethod
def build() -> dict:
raise NotImplementedError('Build method is not defined!')
class SingleShot(InterventionConstructor):
"""
Construct single-shot interventions
"""
@staticmethod
def build(**kwargs) -> dict:
"""
Construct a single shot intervention case
Parameters
----------
state_vector: dict
keys --> concepts, values --> state of the concept.
weight_matrix: pd.DataFrame
causal weights between concepts
Return
----------
y: dictionary
"""
intervention = {}
intervention['state_vector'] = kwargs['initial_state']
intervention['weight_matrix'] = kwargs['weight_matrix']
return intervention
class Continuous(InterventionConstructor):
@staticmethod
def build(**kwargs) -> dict:
"""
Construct a continuous intervention case
Parameters
----------
weight_matrix: pd.DataFrame
causal weights between concepts
initial_state: dict
keys --> concepts, values --> state of the concept.
equilibriums: dict
keys --> concepts, values --> equilibrium states of the concept.
impact: dict
keys ---> concepts the intervention impacts, value: the associated causal weight
effectiveness: float
the degree to which the intervention was delivered (should be between [-1, 1])
default --> 1
Return
----------
y: dictionary
"""
weight_matrix = kwargs['weight_matrix']
initial_state = kwargs['initial_state']
equilibriums = kwargs['equilibriums']
impact = kwargs['params']['impact']
# Set the intervention effectiveness to 1 if the optional parameter is not specified.
try:
effectiveness = kwargs['params']['effectiveness']
except:
effectiveness = 1
# Check whether the passed intervention inputs are in the function's domain.
if (min(list(impact.values())) < -1) or (max(list(impact.values())) > 1):
raise ValueError('the values in the causal weights are out of the domain [-1,1].')
elif (effectiveness < 0) or (effectiveness > 1):
raise ValueError('the values in the intervention effectiveness are out of the domain [0,1].')
intervention = {}
intervention['effectiveness'] = effectiveness
# construct a weight matrix for a given intervention
if type(weight_matrix) == np.ndarray:
temp = pd.DataFrame(weight_matrix, columns=initial_state)
else:
temp = weight_matrix.copy(deep=True)
temp['antecedent'] = temp.columns
temp.set_index('antecedent', inplace=True)
temp['intervention'] = 0
temp.loc[len(temp)] = 0
temp.rename(index = {temp.index[-1] : 'intervention'}, inplace = True)
# add the intervention impact
for key in impact.keys():
temp.loc['intervention', key] = impact[key]
# construct the new state vector for a given intervention (baseline + intervention effectiveness)
temp_vector = equilibriums['baseline'].copy(deep=True)
temp_vector = temp_vector.append(pd.Series({'intervention': effectiveness})).to_dict()
# add the causal weights for the intervention
intervention['weight_matrix'] = temp
intervention['state_vector'] = temp_vector
return intervention
Classes
class Continuous
-
Class of methods for building interventions.
Expand source code
class Continuous(InterventionConstructor): @staticmethod def build(**kwargs) -> dict: """ Construct a continuous intervention case Parameters ---------- weight_matrix: pd.DataFrame causal weights between concepts initial_state: dict keys --> concepts, values --> state of the concept. equilibriums: dict keys --> concepts, values --> equilibrium states of the concept. impact: dict keys ---> concepts the intervention impacts, value: the associated causal weight effectiveness: float the degree to which the intervention was delivered (should be between [-1, 1]) default --> 1 Return ---------- y: dictionary """ weight_matrix = kwargs['weight_matrix'] initial_state = kwargs['initial_state'] equilibriums = kwargs['equilibriums'] impact = kwargs['params']['impact'] # Set the intervention effectiveness to 1 if the optional parameter is not specified. try: effectiveness = kwargs['params']['effectiveness'] except: effectiveness = 1 # Check whether the passed intervention inputs are in the function's domain. if (min(list(impact.values())) < -1) or (max(list(impact.values())) > 1): raise ValueError('the values in the causal weights are out of the domain [-1,1].') elif (effectiveness < 0) or (effectiveness > 1): raise ValueError('the values in the intervention effectiveness are out of the domain [0,1].') intervention = {} intervention['effectiveness'] = effectiveness # construct a weight matrix for a given intervention if type(weight_matrix) == np.ndarray: temp = pd.DataFrame(weight_matrix, columns=initial_state) else: temp = weight_matrix.copy(deep=True) temp['antecedent'] = temp.columns temp.set_index('antecedent', inplace=True) temp['intervention'] = 0 temp.loc[len(temp)] = 0 temp.rename(index = {temp.index[-1] : 'intervention'}, inplace = True) # add the intervention impact for key in impact.keys(): temp.loc['intervention', key] = impact[key] # construct the new state vector for a given intervention (baseline + intervention effectiveness) temp_vector = equilibriums['baseline'].copy(deep=True) temp_vector = temp_vector.append(pd.Series({'intervention': effectiveness})).to_dict() # add the causal weights for the intervention intervention['weight_matrix'] = temp intervention['state_vector'] = temp_vector return intervention
Ancestors
- InterventionConstructor
- abc.ABC
Static methods
def build(**kwargs) ‑> dict
-
Construct a continuous intervention case
Parameters
weight_matrix
:pd.DataFrame
- causal weights between concepts
initial_state
:dict
- keys –> concepts, values –> state of the concept.
equilibriums
:dict
- keys –> concepts, values –> equilibrium states of the concept.
impact
:dict
- keys —> concepts the intervention impacts, value: the associated causal weight
effectiveness
:float
- the degree to which the intervention was delivered (should be between [-1, 1]) default –> 1
Return
y: dictionary
Expand source code
@staticmethod def build(**kwargs) -> dict: """ Construct a continuous intervention case Parameters ---------- weight_matrix: pd.DataFrame causal weights between concepts initial_state: dict keys --> concepts, values --> state of the concept. equilibriums: dict keys --> concepts, values --> equilibrium states of the concept. impact: dict keys ---> concepts the intervention impacts, value: the associated causal weight effectiveness: float the degree to which the intervention was delivered (should be between [-1, 1]) default --> 1 Return ---------- y: dictionary """ weight_matrix = kwargs['weight_matrix'] initial_state = kwargs['initial_state'] equilibriums = kwargs['equilibriums'] impact = kwargs['params']['impact'] # Set the intervention effectiveness to 1 if the optional parameter is not specified. try: effectiveness = kwargs['params']['effectiveness'] except: effectiveness = 1 # Check whether the passed intervention inputs are in the function's domain. if (min(list(impact.values())) < -1) or (max(list(impact.values())) > 1): raise ValueError('the values in the causal weights are out of the domain [-1,1].') elif (effectiveness < 0) or (effectiveness > 1): raise ValueError('the values in the intervention effectiveness are out of the domain [0,1].') intervention = {} intervention['effectiveness'] = effectiveness # construct a weight matrix for a given intervention if type(weight_matrix) == np.ndarray: temp = pd.DataFrame(weight_matrix, columns=initial_state) else: temp = weight_matrix.copy(deep=True) temp['antecedent'] = temp.columns temp.set_index('antecedent', inplace=True) temp['intervention'] = 0 temp.loc[len(temp)] = 0 temp.rename(index = {temp.index[-1] : 'intervention'}, inplace = True) # add the intervention impact for key in impact.keys(): temp.loc['intervention', key] = impact[key] # construct the new state vector for a given intervention (baseline + intervention effectiveness) temp_vector = equilibriums['baseline'].copy(deep=True) temp_vector = temp_vector.append(pd.Series({'intervention': effectiveness})).to_dict() # add the causal weights for the intervention intervention['weight_matrix'] = temp intervention['state_vector'] = temp_vector return intervention
class InterventionConstructor
-
Class of methods for building interventions.
Expand source code
class InterventionConstructor(ABC): """ Class of methods for building interventions. """ @abstractmethod def build() -> dict: raise NotImplementedError('Build method is not defined!')
Ancestors
- abc.ABC
Subclasses
Methods
def build() ‑> dict
-
Expand source code
@abstractmethod def build() -> dict: raise NotImplementedError('Build method is not defined!')
class SingleShot
-
Construct single-shot interventions
Expand source code
class SingleShot(InterventionConstructor): """ Construct single-shot interventions """ @staticmethod def build(**kwargs) -> dict: """ Construct a single shot intervention case Parameters ---------- state_vector: dict keys --> concepts, values --> state of the concept. weight_matrix: pd.DataFrame causal weights between concepts Return ---------- y: dictionary """ intervention = {} intervention['state_vector'] = kwargs['initial_state'] intervention['weight_matrix'] = kwargs['weight_matrix'] return intervention
Ancestors
- InterventionConstructor
- abc.ABC
Static methods
def build(**kwargs) ‑> dict
-
Construct a single shot intervention case
Parameters
state_vector
:dict
- keys –> concepts, values –> state of the concept.
weight_matrix
:pd.DataFrame
- causal weights between concepts
Return
y: dictionary
Expand source code
@staticmethod def build(**kwargs) -> dict: """ Construct a single shot intervention case Parameters ---------- state_vector: dict keys --> concepts, values --> state of the concept. weight_matrix: pd.DataFrame causal weights between concepts Return ---------- y: dictionary """ intervention = {} intervention['state_vector'] = kwargs['initial_state'] intervention['weight_matrix'] = kwargs['weight_matrix'] return intervention