Merge pull request #166 from keli6/feature/TRXF-PMML-ScoreCard-export

TRXF pmml scorecard writer

Author: vijay-arya

aix360/algorithms/rule_induction/trxf/pmml_export/models/__init__.py

@@ -2,17 +2,23 @@
 from .data_dictionary import DataField
 from .data_dictionary import DataType
 from .data_dictionary import OpType
+from .data_dictionary import Value
+from .data_dictionary import Restriction
 from .mining_schema import MiningField
 from .mining_schema import MiningFieldUsageType
 from .mining_schema import MiningSchema
 from .predicate import BooleanOperator
 from .predicate import CompoundPredicate
 from .predicate import Operator
 from .predicate import SimplePredicate
+from .predicate import TruePredicate
 from .rule import SimpleRule
 from .rule import DEFAULT_WEIGHT
 from .rule import DEFAULT_CONFIDENCE
 from .ruleset import RuleSelectionMethod
 from .ruleset import RuleSet
 from .ruleset_model import RuleSetModel
 from .pmml_ruleset_model import SimplePMMLRuleSetModel
+from .scorecard import Scorecard, Output, OutputField
+from .characteristics import Characteristics, Characteristic
+from .attribute import Attribute, ComplexPartialScore

aix360/algorithms/rule_induction/trxf/pmml_export/models/attribute.py

@@ -0,0 +1,24 @@
+import typing
+
+from . import predicate
+
+try:
+    # python >= 3.7
+    from dataclasses import dataclass as dataclass
+    from dataclasses import field as field
+except ImportError:
+    from attr import s as dataclass
+    from attr import ib as field
+
+
+@dataclass(frozen=True)
+class ComplexPartialScore:
+    feature_name: str = field()
+    multiplier: str = field(default=None)
+    constant: str = field(default=None)
+
+
+@dataclass(frozen=True)
+class Attribute:
+    score: typing.Union[str, ComplexPartialScore]
+    predicate: typing.Union[predicate.SimplePredicate, predicate.CompoundPredicate, predicate.TruePredicate]

aix360/algorithms/rule_induction/trxf/pmml_export/models/characteristics.py

@@ -0,0 +1,20 @@
+import typing
+
+from . import attribute
+
+try:
+    # python >= 3.7
+    from dataclasses import dataclass as dataclass
+except ImportError:
+    from attr import s as dataclass
+
+
+@dataclass(frozen=True)
+class Characteristic:
+    name: str
+    attributes: typing.List[attribute.Attribute]
+
+
+@dataclass(frozen=True)
+class Characteristics:
+    characteristics: typing.List[Characteristic]

aix360/algorithms/rule_induction/trxf/pmml_export/models/data_dictionary.py

@@ -23,9 +23,16 @@ class OpType(enum.Enum):
     continuous = 2
 
 
+class Restriction(enum.Enum):
+    valid = 0
+    invalid = 1
+    missing = 2
+
+
 @dataclass(frozen=True)
 class Value:
     value: str = field()
+    property: Restriction = field(default=Restriction.valid)
 
 
 @dataclass(frozen=True)
@@ -35,6 +42,11 @@ class DataField:
     dataType: DataType = field()
     values: typing.Optional[typing.List[Value]] = field(default=None)
 
+    def __post_init__(self):
+        if self.values and \
+                (self.dataType is not DataType.string or self.optype not in (OpType.ordinal, OpType.categorical)):
+            raise ValueError
+
 
 @dataclass(frozen=True)
 class DataDictionary:

aix360/algorithms/rule_induction/trxf/pmml_export/models/predicate.py

@@ -16,13 +16,19 @@ class Operator(enum.Enum):
     lessOrEqual = 3
     greaterThan = 4
     greaterOrEqual = 5
+    isMissing = 6
 
 
 @dataclass(frozen=True)
 class SimplePredicate:
     operator: Operator = field()
-    value: str = field()
     field: str = field()
+    value: typing.Optional[str] = None
+
+
+@dataclass(frozen=True)
+class TruePredicate:
+    pass
 
 
 # Use functional api to add aliases for `or` and `and`

aix360/algorithms/rule_induction/trxf/pmml_export/models/scorecard.py

@@ -0,0 +1,35 @@
+import typing
+
+from . import characteristics
+from . import data_dictionary
+from . import mining_schema
+
+try:
+    # python >= 3.7
+    from dataclasses import dataclass as dataclass
+    from dataclasses import field as field
+except ImportError:
+    from attr import s as dataclass
+    from attr import ib as field
+
+
+@dataclass(frozen=True)
+class OutputField:
+    name: str
+    feature: str
+    dataType: data_dictionary.DataType
+    optype: data_dictionary.OpType
+
+
+@dataclass(frozen=True)
+class Output:
+    outputFields: typing.List[OutputField]
+
+
+@dataclass(frozen=True)
+class Scorecard:
+    dataDictionary: data_dictionary.DataDictionary
+    miningSchema: mining_schema.MiningSchema
+    output: Output
+    characteristics: characteristics.Characteristics
+    initialScore: str = field(default="0")

aix360/algorithms/rule_induction/trxf/pmml_export/serializer/abstract_serializer.py

@@ -1,10 +1,11 @@
 import abc
+import typing
 
 from .. import models
 
 
 class AbstractSerializer(abc.ABC):
 
     @abc.abstractmethod
-    def serialize(self, simple_pmml_ruleset_model: models.SimplePMMLRuleSetModel) -> str:
+    def serialize(self, model: typing.Union[models.SimplePMMLRuleSetModel, models.Scorecard]) -> str:
         raise NotImplementedError

aix360/algorithms/rule_induction/trxf/pmml_export/serializer/nyoka_serializer.py

@@ -1,3 +1,4 @@
+import typing
 import datetime
 import io
 
@@ -16,36 +17,49 @@ class NyokaSerializer(AbstractSerializer):
     def __init__(self, timestamp: datetime = None):
         self._timestamp = timestamp
 
-    def serialize(self, simple_pmml_ruleset_model: models.SimplePMMLRuleSetModel, timestamp: datetime = None) -> str:
-        pmml_model = self._nyoka_pmml_model(simple_pmml_ruleset_model)
+    def serialize(self, model: typing.Union[models.SimplePMMLRuleSetModel, models.Scorecard]) -> str:
+        if isinstance(model, models.SimplePMMLRuleSetModel):
+            pmml_model = self._nyoka_pmml_model(
+                model,
+                RuleSetModel=None if model.ruleSetModel is None else [self._nyoka_rule_set_model(model.ruleSetModel)])
+        elif isinstance(model, models.Scorecard):
+            pmml_model = self._nyoka_pmml_model(
+                model,
+                Scorecard=None if model is None else [self._nyoka_scorecard_model(model)])
+        else:
+            raise NotImplemented
         string_io = io.StringIO()
         pmml_model.export(outfile=string_io, level=0)
         return string_io.getvalue()
 
-    def _nyoka_pmml_model(self, simple_pmml_ruleset_model: models.SimplePMMLRuleSetModel) -> nyoka_pmml.PMML:
-        timestamp = datetime.datetime.now() if self._timestamp is None else self._timestamp
+    def _nyoka_pmml_model(
+            self,
+            model: typing.Union[models.SimplePMMLRuleSetModel, models.Scorecard],
+            **build_args
+    ) -> nyoka_pmml.PMML:
         return nyoka_pmml.PMML(
             version=nyoka_constants.PMML_SCHEMA.VERSION,
             Header=nyoka_pmml.Header(
                 copyright=NyokaSerializer.COPYRIGHT_STRING,
                 description=nyoka_constants.HEADER_INFO.DEFAULT_DESCRIPTION,
-                Timestamp=nyoka_pmml.Timestamp(timestamp),
-                Application=nyoka_pmml.Application(
-                    name=NyokaSerializer.APPLICATION_NAME, version=version.version)),
-            DataDictionary=None if simple_pmml_ruleset_model.dataDictionary is None else self._nyoka_data_dictionary(
-                simple_pmml_ruleset_model.dataDictionary),
-            RuleSetModel=None if simple_pmml_ruleset_model.ruleSetModel is None else [
-                self._nyoka_rule_set_model(simple_pmml_ruleset_model.ruleSetModel)])
+                Timestamp=nyoka_pmml.Timestamp(datetime.datetime.now() if self._timestamp is None else self._timestamp),
+                Application=nyoka_pmml.Application(name=NyokaSerializer.APPLICATION_NAME, version=version.version)),
+            DataDictionary=None if model.dataDictionary is None else self._nyoka_data_dictionary(model.dataDictionary),
+            **build_args)
 
     def _nyoka_data_dictionary(self, data_dictionary: models.DataDictionary) -> nyoka_pmml.DataDictionary:
         return nyoka_pmml.DataDictionary(
-            numberOfFields=0 if data_dictionary.dataFields is None else len(data_dictionary.dataFields),
-            DataField=None if data_dictionary.dataFields is None else [
-                nyoka_pmml.DataField(name=f.name, optype=f.optype.name, dataType=f.dataType.name,
-                                     Value=None if f.values is None else [
-                                         nyoka_pmml.Value(value=v.value) for v in f.values
-                                     ])
-                for f in data_dictionary.dataFields])
+            numberOfFields=0 if not data_dictionary.dataFields else len(data_dictionary.dataFields),
+            DataField=None if not data_dictionary.dataFields else [
+                self._nyoka_data_field(f) for f in data_dictionary.dataFields])
+
+    def _nyoka_data_field(self, data_field: models.DataField) -> nyoka_pmml.DataField:
+        return nyoka_pmml.DataField(
+            name=data_field.name,
+            optype=data_field.optype.name,
+            dataType=data_field.dataType.name,
+            Value=None if not data_field.values else [
+                nyoka_pmml.Value(value=val.value, property=val.property.name) for val in data_field.values])
 
     def _nyoka_rule_set_model(self, rule_set_model: models.RuleSetModel) -> nyoka_pmml.RuleSetModel:
         return nyoka_pmml.RuleSetModel(
@@ -94,3 +108,55 @@ def _nyoka_rule(self, simple_rule: models.SimpleRule) -> nyoka_pmml.SimpleRule:
             nbCorrect=simple_rule.nbCorrect,
             confidence=simple_rule.confidence,
             weight=simple_rule.weight)
+
+    def _nyoka_scorecard_model(self, scorecard: models.Scorecard) -> nyoka_pmml.Scorecard:
+        return nyoka_pmml.Scorecard(
+            functionName='regression',
+            algorithmName='ScoreCard',
+            MiningSchema=None if scorecard.miningSchema is None else self._nyoka_mining_schema(
+                scorecard.miningSchema),
+            initialScore=scorecard.initialScore,
+            useReasonCodes="false",
+            Output=None if scorecard.output is None else nyoka_pmml.Output(
+                OutputField=[
+                    nyoka_pmml.OutputField(
+                        name=outputField.name,
+                        feature=outputField.feature,
+                        dataType=outputField.dataType.name,
+                        optype=outputField.optype.name) for outputField in scorecard.output.outputFields]),
+            Characteristics=None if scorecard.characteristics is None else self._nyoka_pmml_characteristics(
+                scorecard.characteristics))
+
+    def _nyoka_pmml_characteristics(self, characteristics: models.Characteristics) -> nyoka_pmml.Characteristics:
+        return nyoka_pmml.Characteristics(
+            Characteristic=[
+                nyoka_pmml.Characteristic(
+                    name=characteristic.name,
+                    Attribute=[self._nyoka_pmml_attributes(attribute) for attribute in characteristic.attributes])
+                for characteristic in characteristics.characteristics])
+
+    def _nyoka_pmml_attributes(self, attribute: models.Attribute) -> nyoka_pmml.Attribute:
+        return nyoka_pmml.Attribute(
+            partialScore=attribute.score if not isinstance(attribute.score, models.ComplexPartialScore) else None,
+            ComplexPartialScore=nyoka_pmml.ComplexPartialScore(
+                Apply=nyoka_pmml.Apply(
+                    function='+',
+                    Apply_member=[nyoka_pmml.Apply(
+                        function='*',
+                        FieldRef=[nyoka_pmml.FieldRef(field=attribute.score.feature_name)],
+                        Constant=[nyoka_pmml.Constant(valueOf_=attribute.score.multiplier)])],
+                    Constant=[nyoka_pmml.Constant(valueOf_=attribute.score.constant)])) if isinstance(
+                        attribute.score, models.ComplexPartialScore) else None,
+            SimplePredicate=None if (attribute.predicate is None or not isinstance(
+                attribute.predicate, models.SimplePredicate)) else nyoka_pmml.SimplePredicate(
+                    field=attribute.predicate.field,
+                    operator=attribute.predicate.operator.name,
+                    value=attribute.predicate.value),
+            CompoundPredicate=None if (attribute.predicate is None or not isinstance(
+                attribute.predicate, models.CompoundPredicate)) else nyoka_pmml.CompoundPredicate(
+                    booleanOperator=attribute.predicate.booleanOperator.name,
+                    SimplePredicate=[
+                        nyoka_pmml.SimplePredicate(field=sp.field, operator=sp.operator.name, value=sp.value)
+                        for sp in attribute.predicate.simplePredicates]),
+            True_=None if (attribute.predicate is None or not isinstance(
+                attribute.predicate, models.TruePredicate)) else nyoka_pmml.True_())