samoa-api/src/main/java/com/yahoo/labs/samoa/learners/classifiers/rules/centralized/AMRulesRegressorProcessor.java - incubator-samoa - Git at Google

 package com.yahoo.labs.samoa.learners.classifiers.rules.centralized;

 /*
  * #%L
  * SAMOA
  * %%
  * Copyright (C) 2013 - 2014 Yahoo! Inc.
  * %%
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  * #L%
  */

 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;

 import com.yahoo.labs.samoa.core.ContentEvent;
 import com.yahoo.labs.samoa.core.Processor;
 import com.yahoo.labs.samoa.instances.Instance;
 import com.yahoo.labs.samoa.instances.Instances;
 import com.yahoo.labs.samoa.learners.InstanceContentEvent;
 import com.yahoo.labs.samoa.learners.ResultContentEvent;
 import com.yahoo.labs.samoa.learners.classifiers.rules.common.ActiveRule;
 import com.yahoo.labs.samoa.learners.classifiers.rules.common.Perceptron;
 import com.yahoo.labs.samoa.learners.classifiers.rules.common.RuleActiveRegressionNode;
 import com.yahoo.labs.samoa.moa.classifiers.rules.core.attributeclassobservers.FIMTDDNumericAttributeClassLimitObserver;
 import com.yahoo.labs.samoa.moa.classifiers.rules.core.voting.ErrorWeightedVote;
 import com.yahoo.labs.samoa.topology.Stream;

 /**
  * AMRules Regressor Processor
  * is the main (and only) processor for AMRulesRegressor task.
  * It is adapted from the AMRules implementation in MOA.
  *
  * @author Anh Thu Vu
  *
  */
 public class AMRulesRegressorProcessor implements Processor {
 	/**
 	 *
 	 */
 	private static final long serialVersionUID = 1L;

 	private int processorId;

 	// Rules & default rule
 	protected List<ActiveRule> ruleSet;
 	protected ActiveRule defaultRule;
 	protected int ruleNumberID;
 	protected double[] statistics;

 	// SAMOA Stream
 	private Stream resultStream;

 	// Options
 	protected int pageHinckleyThreshold;
 	protected double pageHinckleyAlpha;
 	protected boolean driftDetection;
 	protected int predictionFunction; // Adaptive=0 Perceptron=1 TargetMean=2
 	protected boolean constantLearningRatioDecay;
 	protected double learningRatio;

 	protected double splitConfidence;
 	protected double tieThreshold;
 	protected int gracePeriod;

 	protected boolean noAnomalyDetection;
 	protected double multivariateAnomalyProbabilityThreshold;
 	protected double univariateAnomalyprobabilityThreshold;
 	protected int anomalyNumInstThreshold;

 	protected boolean unorderedRules;

 	protected FIMTDDNumericAttributeClassLimitObserver numericObserver;

 	protected ErrorWeightedVote voteType;

 	/*
 	 * Constructor
 	 */
 	public AMRulesRegressorProcessor (Builder builder) {
 		this.pageHinckleyThreshold = builder.pageHinckleyThreshold;
 		this.pageHinckleyAlpha = builder.pageHinckleyAlpha;
 		this.driftDetection = builder.driftDetection;
 		this.predictionFunction = builder.predictionFunction;
 		this.constantLearningRatioDecay = builder.constantLearningRatioDecay;
 		this.learningRatio = builder.learningRatio;
 		this.splitConfidence = builder.splitConfidence;
 		this.tieThreshold = builder.tieThreshold;
 		this.gracePeriod = builder.gracePeriod;

 		this.noAnomalyDetection = builder.noAnomalyDetection;
 		this.multivariateAnomalyProbabilityThreshold = builder.multivariateAnomalyProbabilityThreshold;
 		this.univariateAnomalyprobabilityThreshold = builder.univariateAnomalyprobabilityThreshold;
 		this.anomalyNumInstThreshold = builder.anomalyNumInstThreshold;
 		this.unorderedRules = builder.unorderedRules;

 		this.numericObserver = builder.numericObserver;
 		this.voteType = builder.voteType;
 	}

 	/*
 	 * Process
 	 */
 	@Override
 	public boolean process(ContentEvent event) {
 		InstanceContentEvent instanceEvent = (InstanceContentEvent) event;

 		// predict
 		if (instanceEvent.isTesting()) {
 			this.predictOnInstance(instanceEvent);
 		}

 		// train
 		if (instanceEvent.isTraining()) {
 			this.trainOnInstance(instanceEvent);
 		}

 		return true;
 	}

 	/*
 	 * Prediction
 	 */
 	private void predictOnInstance (InstanceContentEvent instanceEvent) {
 		double[] prediction = getVotesForInstance(instanceEvent.getInstance());
 		ResultContentEvent rce = newResultContentEvent(prediction, instanceEvent);
 		resultStream.put(rce);
 	}

 	/**
 	 * Helper method to generate new ResultContentEvent based on an instance and
 	 * its prediction result.
 	 * @param prediction The predicted class label from the decision tree model.
 	 * @param inEvent The associated instance content event
 	 * @return ResultContentEvent to be sent into Evaluator PI or other destination PI.
 	 */
 	private ResultContentEvent newResultContentEvent(double[] prediction, InstanceContentEvent inEvent){
 		ResultContentEvent rce = new ResultContentEvent(inEvent.getInstanceIndex(), inEvent.getInstance(), inEvent.getClassId(), prediction, inEvent.isLastEvent());
 		rce.setClassifierIndex(this.processorId);
 		rce.setEvaluationIndex(inEvent.getEvaluationIndex());
 		return rce;
 	}

 	/**
 	 * getVotesForInstance extension of the instance method getVotesForInstance
 	 * in moa.classifier.java
 	 * returns the prediction of the instance.
 	 * Called in EvaluateModelRegression
 	 */
 	private double[] getVotesForInstance(Instance instance) {
 		ErrorWeightedVote errorWeightedVote=newErrorWeightedVote();
 		int numberOfRulesCovering = 0;

 		for (ActiveRule rule: ruleSet) {
 			if (rule.isCovering(instance) == true){
 				numberOfRulesCovering++;
 				double [] vote=rule.getPrediction(instance);
 				double error= rule.getCurrentError();
 				errorWeightedVote.addVote(vote,error);
 				if (!this.unorderedRules) { // Ordered Rules Option.
 					break; // Only one rule cover the instance.
 				}
 			}
 		}

 		if (numberOfRulesCovering == 0) {
 			double [] vote=defaultRule.getPrediction(instance);
 			double error= defaultRule.getCurrentError();
 			errorWeightedVote.addVote(vote,error);
 		}
 		double[] weightedVote=errorWeightedVote.computeWeightedVote();

 		return weightedVote;
 	}

 	public ErrorWeightedVote newErrorWeightedVote() {
 		return voteType.getACopy();
 	}

 	/*
 	 * Training
 	 */
 	private void trainOnInstance (InstanceContentEvent instanceEvent) {
 		this.trainOnInstanceImpl(instanceEvent.getInstance());
 	}
 	public void trainOnInstanceImpl(Instance instance) {
 		/**
 		 * AMRules Algorithm
 		 *
 		 //For each rule in the rule set
 			//If rule covers the instance
 				//if the instance is not an anomaly
 					//Update Change Detection Tests
 				    	//Compute prediction error
 				    	//Call PHTest
 						//If change is detected then
 							//Remove rule
 						//Else
 							//Update sufficient statistics of rule
 							//If number of examples in rule  > Nmin
 								//Expand rule
 						//If ordered set then
 							//break
 			//If none of the rule covers the instance
 				//Update sufficient statistics of default rule
 				//If number of examples in default rule is multiple of Nmin
 					//Expand default rule and add it to the set of rules
 					//Reset the default rule
 		 */
 		boolean rulesCoveringInstance = false;
 		Iterator<ActiveRule> ruleIterator= this.ruleSet.iterator();
 		while (ruleIterator.hasNext()) {
 			ActiveRule rule = ruleIterator.next();
 			if (rule.isCovering(instance) == true) {
 				rulesCoveringInstance = true;
 				if (isAnomaly(instance, rule) == false) {
 					//Update Change Detection Tests
 					double error = rule.computeError(instance); //Use adaptive mode error
 					boolean changeDetected = ((RuleActiveRegressionNode)rule.getLearningNode()).updateChangeDetection(error);
 					if (changeDetected == true) {
 						ruleIterator.remove();
 					} else {
 						rule.updateStatistics(instance);
 						if (rule.getInstancesSeen()  % this.gracePeriod == 0.0) {
 							if (rule.tryToExpand(this.splitConfidence, this.tieThreshold) ) {
 								rule.split();
 							}
 						}
 					}
 					if (!this.unorderedRules)
 						break;
 				}
 			}
 		}

 		if (rulesCoveringInstance == false){
 			defaultRule.updateStatistics(instance);
 			if (defaultRule.getInstancesSeen() % this.gracePeriod == 0.0) {
 				if (defaultRule.tryToExpand(this.splitConfidence, this.tieThreshold) == true) {
 					ActiveRule newDefaultRule=newRule(defaultRule.getRuleNumberID(),(RuleActiveRegressionNode)defaultRule.getLearningNode(),
 							((RuleActiveRegressionNode)defaultRule.getLearningNode()).getStatisticsOtherBranchSplit()); //other branch
 					defaultRule.split();
 					defaultRule.setRuleNumberID(++ruleNumberID);
 					this.ruleSet.add(this.defaultRule);

 					defaultRule=newDefaultRule;

 				}
 			}
 		}
 	}

 	/**
 	 * Method to verify if the instance is an anomaly.
 	 * @param instance
 	 * @param rule
 	 * @return
 	 */
 	private boolean isAnomaly(Instance instance, ActiveRule rule) {
 		//AMRUles is equipped with anomaly detection. If on, compute the anomaly value.
 		boolean isAnomaly = false;
 		if (this.noAnomalyDetection == false){
 			if (rule.getInstancesSeen() >= this.anomalyNumInstThreshold) {
 				isAnomaly = rule.isAnomaly(instance,
 						this.univariateAnomalyprobabilityThreshold,
 						this.multivariateAnomalyProbabilityThreshold,
 						this.anomalyNumInstThreshold);
 			}
 		}
 		return isAnomaly;
 	}

 	/*
 	 * Create new rules
 	 */
 	// TODO check this after finish rule, LN
 	private ActiveRule newRule(int ID, RuleActiveRegressionNode node, double[] statistics) {
 		ActiveRule r=newRule(ID);

 		if (node!=null)
 		{
 			if(node.getPerceptron()!=null)
 			{
 				r.getLearningNode().setPerceptron(new Perceptron(node.getPerceptron()));
 				r.getLearningNode().getPerceptron().setLearningRatio(this.learningRatio);
 			}
 			if (statistics==null)
 			{
 				double mean;
 				if(node.getNodeStatistics().getValue(0)>0){
 					mean=node.getNodeStatistics().getValue(1)/node.getNodeStatistics().getValue(0);
 					r.getLearningNode().getTargetMean().reset(mean, 1);
 				}
 			}
 		}
 		if (statistics!=null && ((RuleActiveRegressionNode)r.getLearningNode()).getTargetMean()!=null)
 		{
 			double mean;
 			if(statistics[0]>0){
 				mean=statistics[1]/statistics[0];
 				((RuleActiveRegressionNode)r.getLearningNode()).getTargetMean().reset(mean, (long)statistics[0]);
 			}
 		}
 		return r;
 	}

 	private ActiveRule newRule(int ID) {
 		ActiveRule r=new ActiveRule.Builder().
 				threshold(this.pageHinckleyThreshold).
 				alpha(this.pageHinckleyAlpha).
 				changeDetection(this.driftDetection).
 				predictionFunction(this.predictionFunction).
 				statistics(new double[3]).
 				learningRatio(this.learningRatio).
 				numericObserver(numericObserver).
 				id(ID).build();
 		return r;
 	}

 	/*
 	 * Init processor
 	 */
 	@Override
 	public void onCreate(int id) {
 		this.processorId = id;
 		this.statistics= new double[]{0.0,0,0};
 		this.ruleNumberID=0;
 		this.defaultRule = newRule(++this.ruleNumberID);

 		this.ruleSet = new LinkedList<ActiveRule>();
 	}

 	/*
 	 * Clone processor
 	 */
 	@Override
 	public Processor newProcessor(Processor p) {
 		AMRulesRegressorProcessor oldProcessor = (AMRulesRegressorProcessor) p;
 		Builder builder = new Builder(oldProcessor);
 		AMRulesRegressorProcessor newProcessor = builder.build();
 		newProcessor.resultStream = oldProcessor.resultStream;
 		return newProcessor;
 	}

 	/*
 	 * Output stream
 	 */
 	public void setResultStream(Stream resultStream) {
 		this.resultStream = resultStream;
 	}

 	public Stream getResultStream() {
 		return this.resultStream;
 	}

 	/*
 	 * Others
 	 */
     public boolean isRandomizable() {
     	return true;
     }

 	/*
 	 * Builder
 	 */
 	public static class Builder {
 		private int pageHinckleyThreshold;
 		private double pageHinckleyAlpha;
 		private boolean driftDetection;
 		private int predictionFunction; // Adaptive=0 Perceptron=1 TargetMean=2
 		private boolean constantLearningRatioDecay;
 		private double learningRatio;
 		private double splitConfidence;
 		private double tieThreshold;
 		private int gracePeriod;

 		private boolean noAnomalyDetection;
 		private double multivariateAnomalyProbabilityThreshold;
 		private double univariateAnomalyprobabilityThreshold;
 		private int anomalyNumInstThreshold;

 		private boolean unorderedRules;

 		private FIMTDDNumericAttributeClassLimitObserver numericObserver;
 		private ErrorWeightedVote voteType;

 		private Instances dataset;

 		public Builder(Instances dataset){
 			this.dataset = dataset;
 		}

 		public Builder(AMRulesRegressorProcessor processor) {
 			this.pageHinckleyThreshold = processor.pageHinckleyThreshold;
 			this.pageHinckleyAlpha = processor.pageHinckleyAlpha;
 			this.driftDetection = processor.driftDetection;
 			this.predictionFunction = processor.predictionFunction;
 			this.constantLearningRatioDecay = processor.constantLearningRatioDecay;
 			this.learningRatio = processor.learningRatio;
 			this.splitConfidence = processor.splitConfidence;
 			this.tieThreshold = processor.tieThreshold;
 			this.gracePeriod = processor.gracePeriod;

 			this.noAnomalyDetection = processor.noAnomalyDetection;
 			this.multivariateAnomalyProbabilityThreshold = processor.multivariateAnomalyProbabilityThreshold;
 			this.univariateAnomalyprobabilityThreshold = processor.univariateAnomalyprobabilityThreshold;
 			this.anomalyNumInstThreshold = processor.anomalyNumInstThreshold;
 			this.unorderedRules = processor.unorderedRules;

 			this.numericObserver = processor.numericObserver;
 			this.voteType = processor.voteType;
 		}

 		public Builder threshold(int threshold) {
 			this.pageHinckleyThreshold = threshold;
 			return this;
 		}

 		public Builder alpha(double alpha) {
 			this.pageHinckleyAlpha = alpha;
 			return this;
 		}

 		public Builder changeDetection(boolean changeDetection) {
 			this.driftDetection = changeDetection;
 			return this;
 		}

 		public Builder predictionFunction(int predictionFunction) {
 			this.predictionFunction = predictionFunction;
 			return this;
 		}

 		public Builder constantLearningRatioDecay(boolean constantDecay) {
 			this.constantLearningRatioDecay = constantDecay;
 			return this;
 		}

 		public Builder learningRatio(double learningRatio) {
 			this.learningRatio = learningRatio;
 			return this;
 		}

 		public Builder splitConfidence(double splitConfidence) {
 			this.splitConfidence = splitConfidence;
 			return this;
 		}

 		public Builder tieThreshold(double tieThreshold) {
 			this.tieThreshold = tieThreshold;
 			return this;
 		}

 		public Builder gracePeriod(int gracePeriod) {
 			this.gracePeriod = gracePeriod;
 			return this;
 		}

 		public Builder noAnomalyDetection(boolean noAnomalyDetection) {
 			this.noAnomalyDetection = noAnomalyDetection;
 			return this;
 		}

 		public Builder multivariateAnomalyProbabilityThreshold(double mAnomalyThreshold) {
 			this.multivariateAnomalyProbabilityThreshold = mAnomalyThreshold;
 			return this;
 		}

 		public Builder univariateAnomalyProbabilityThreshold(double uAnomalyThreshold) {
 			this.univariateAnomalyprobabilityThreshold = uAnomalyThreshold;
 			return this;
 		}

 		public Builder anomalyNumberOfInstancesThreshold(int anomalyNumInstThreshold) {
 			this.anomalyNumInstThreshold = anomalyNumInstThreshold;
 			return this;
 		}

 		public Builder unorderedRules(boolean unorderedRules) {
 			this.unorderedRules = unorderedRules;
 			return this;
 		}

 		public Builder numericObserver(FIMTDDNumericAttributeClassLimitObserver numericObserver) {
 			this.numericObserver = numericObserver;
 			return this;
 		}

 		public Builder voteType(ErrorWeightedVote voteType) {
 			this.voteType = voteType;
 			return this;
 		}

 		public AMRulesRegressorProcessor build() {
 			return new AMRulesRegressorProcessor(this);
 		}
 	}
 }
	package com.yahoo.labs.samoa.learners.classifiers.rules.centralized;

	/*
	* #%L
	* SAMOA
	* %%
	* Copyright (C) 2013 - 2014 Yahoo! Inc.
	* %%
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	* #L%
	*/

	import java.util.Iterator;
	import java.util.LinkedList;
	import java.util.List;

	import com.yahoo.labs.samoa.core.ContentEvent;
	import com.yahoo.labs.samoa.core.Processor;
	import com.yahoo.labs.samoa.instances.Instance;
	import com.yahoo.labs.samoa.instances.Instances;
	import com.yahoo.labs.samoa.learners.InstanceContentEvent;
	import com.yahoo.labs.samoa.learners.ResultContentEvent;
	import com.yahoo.labs.samoa.learners.classifiers.rules.common.ActiveRule;
	import com.yahoo.labs.samoa.learners.classifiers.rules.common.Perceptron;
	import com.yahoo.labs.samoa.learners.classifiers.rules.common.RuleActiveRegressionNode;
	import com.yahoo.labs.samoa.moa.classifiers.rules.core.attributeclassobservers.FIMTDDNumericAttributeClassLimitObserver;
	import com.yahoo.labs.samoa.moa.classifiers.rules.core.voting.ErrorWeightedVote;
	import com.yahoo.labs.samoa.topology.Stream;

	/**
	* AMRules Regressor Processor
	* is the main (and only) processor for AMRulesRegressor task.
	* It is adapted from the AMRules implementation in MOA.
	*
	* @author Anh Thu Vu
	*
	*/
	public class AMRulesRegressorProcessor implements Processor {
	/**
	*
	*/
	private static final long serialVersionUID = 1L;

	private int processorId;

	// Rules & default rule
	protected List<ActiveRule> ruleSet;
	protected ActiveRule defaultRule;
	protected int ruleNumberID;
	protected double[] statistics;

	// SAMOA Stream
	private Stream resultStream;

	// Options
	protected int pageHinckleyThreshold;
	protected double pageHinckleyAlpha;
	protected boolean driftDetection;
	protected int predictionFunction; // Adaptive=0 Perceptron=1 TargetMean=2
	protected boolean constantLearningRatioDecay;
	protected double learningRatio;

	protected double splitConfidence;
	protected double tieThreshold;
	protected int gracePeriod;

	protected boolean noAnomalyDetection;
	protected double multivariateAnomalyProbabilityThreshold;
	protected double univariateAnomalyprobabilityThreshold;
	protected int anomalyNumInstThreshold;

	protected boolean unorderedRules;

	protected FIMTDDNumericAttributeClassLimitObserver numericObserver;

	protected ErrorWeightedVote voteType;

	/*
	* Constructor
	*/
	public AMRulesRegressorProcessor (Builder builder) {
	this.pageHinckleyThreshold = builder.pageHinckleyThreshold;
	this.pageHinckleyAlpha = builder.pageHinckleyAlpha;
	this.driftDetection = builder.driftDetection;
	this.predictionFunction = builder.predictionFunction;
	this.constantLearningRatioDecay = builder.constantLearningRatioDecay;
	this.learningRatio = builder.learningRatio;
	this.splitConfidence = builder.splitConfidence;
	this.tieThreshold = builder.tieThreshold;
	this.gracePeriod = builder.gracePeriod;

	this.noAnomalyDetection = builder.noAnomalyDetection;
	this.multivariateAnomalyProbabilityThreshold = builder.multivariateAnomalyProbabilityThreshold;
	this.univariateAnomalyprobabilityThreshold = builder.univariateAnomalyprobabilityThreshold;
	this.anomalyNumInstThreshold = builder.anomalyNumInstThreshold;
	this.unorderedRules = builder.unorderedRules;

	this.numericObserver = builder.numericObserver;
	this.voteType = builder.voteType;
	}

	/*
	* Process
	*/
	@Override
	public boolean process(ContentEvent event) {
	InstanceContentEvent instanceEvent = (InstanceContentEvent) event;

	// predict
	if (instanceEvent.isTesting()) {
	this.predictOnInstance(instanceEvent);
	}

	// train
	if (instanceEvent.isTraining()) {
	this.trainOnInstance(instanceEvent);
	}

	return true;
	}

	/*
	* Prediction
	*/
	private void predictOnInstance (InstanceContentEvent instanceEvent) {
	double[] prediction = getVotesForInstance(instanceEvent.getInstance());
	ResultContentEvent rce = newResultContentEvent(prediction, instanceEvent);
	resultStream.put(rce);
	}

	/**
	* Helper method to generate new ResultContentEvent based on an instance and
	* its prediction result.
	* @param prediction The predicted class label from the decision tree model.
	* @param inEvent The associated instance content event
	* @return ResultContentEvent to be sent into Evaluator PI or other destination PI.
	*/
	private ResultContentEvent newResultContentEvent(double[] prediction, InstanceContentEvent inEvent){
	ResultContentEvent rce = new ResultContentEvent(inEvent.getInstanceIndex(), inEvent.getInstance(), inEvent.getClassId(), prediction, inEvent.isLastEvent());
	rce.setClassifierIndex(this.processorId);
	rce.setEvaluationIndex(inEvent.getEvaluationIndex());
	return rce;
	}

	/**
	* getVotesForInstance extension of the instance method getVotesForInstance
	* in moa.classifier.java
	* returns the prediction of the instance.
	* Called in EvaluateModelRegression
	*/
	private double[] getVotesForInstance(Instance instance) {
	ErrorWeightedVote errorWeightedVote=newErrorWeightedVote();
	int numberOfRulesCovering = 0;

	for (ActiveRule rule: ruleSet) {
	if (rule.isCovering(instance) == true){
	numberOfRulesCovering++;
	double [] vote=rule.getPrediction(instance);
	double error= rule.getCurrentError();
	errorWeightedVote.addVote(vote,error);
	if (!this.unorderedRules) { // Ordered Rules Option.
	break; // Only one rule cover the instance.
	}
	}
	}

	if (numberOfRulesCovering == 0) {
	double [] vote=defaultRule.getPrediction(instance);
	double error= defaultRule.getCurrentError();
	errorWeightedVote.addVote(vote,error);
	}
	double[] weightedVote=errorWeightedVote.computeWeightedVote();

	return weightedVote;
	}

	public ErrorWeightedVote newErrorWeightedVote() {
	return voteType.getACopy();
	}

	/*
	* Training
	*/
	private void trainOnInstance (InstanceContentEvent instanceEvent) {
	this.trainOnInstanceImpl(instanceEvent.getInstance());
	}
	public void trainOnInstanceImpl(Instance instance) {
	/**
	* AMRules Algorithm
	*
	//For each rule in the rule set
	//If rule covers the instance
	//if the instance is not an anomaly
	//Update Change Detection Tests
	//Compute prediction error
	//Call PHTest
	//If change is detected then
	//Remove rule
	//Else
	//Update sufficient statistics of rule
	//If number of examples in rule > Nmin
	//Expand rule
	//If ordered set then
	//break
	//If none of the rule covers the instance
	//Update sufficient statistics of default rule
	//If number of examples in default rule is multiple of Nmin
	//Expand default rule and add it to the set of rules
	//Reset the default rule
	*/
	boolean rulesCoveringInstance = false;
	Iterator<ActiveRule> ruleIterator= this.ruleSet.iterator();
	while (ruleIterator.hasNext()) {
	ActiveRule rule = ruleIterator.next();
	if (rule.isCovering(instance) == true) {
	rulesCoveringInstance = true;
	if (isAnomaly(instance, rule) == false) {
	//Update Change Detection Tests
	double error = rule.computeError(instance); //Use adaptive mode error
	boolean changeDetected = ((RuleActiveRegressionNode)rule.getLearningNode()).updateChangeDetection(error);
	if (changeDetected == true) {
	ruleIterator.remove();
	} else {
	rule.updateStatistics(instance);
	if (rule.getInstancesSeen() % this.gracePeriod == 0.0) {
	if (rule.tryToExpand(this.splitConfidence, this.tieThreshold) ) {
	rule.split();
	}
	}
	}
	if (!this.unorderedRules)
	break;
	}
	}
	}

	if (rulesCoveringInstance == false){
	defaultRule.updateStatistics(instance);
	if (defaultRule.getInstancesSeen() % this.gracePeriod == 0.0) {
	if (defaultRule.tryToExpand(this.splitConfidence, this.tieThreshold) == true) {
	ActiveRule newDefaultRule=newRule(defaultRule.getRuleNumberID(),(RuleActiveRegressionNode)defaultRule.getLearningNode(),
	((RuleActiveRegressionNode)defaultRule.getLearningNode()).getStatisticsOtherBranchSplit()); //other branch
	defaultRule.split();
	defaultRule.setRuleNumberID(++ruleNumberID);
	this.ruleSet.add(this.defaultRule);

	defaultRule=newDefaultRule;

	}
	}
	}
	}

	/**
	* Method to verify if the instance is an anomaly.
	* @param instance
	* @param rule
	* @return
	*/
	private boolean isAnomaly(Instance instance, ActiveRule rule) {
	//AMRUles is equipped with anomaly detection. If on, compute the anomaly value.
	boolean isAnomaly = false;
	if (this.noAnomalyDetection == false){
	if (rule.getInstancesSeen() >= this.anomalyNumInstThreshold) {
	isAnomaly = rule.isAnomaly(instance,
	this.univariateAnomalyprobabilityThreshold,
	this.multivariateAnomalyProbabilityThreshold,
	this.anomalyNumInstThreshold);
	}
	}
	return isAnomaly;
	}

	/*
	* Create new rules
	*/
	// TODO check this after finish rule, LN
	private ActiveRule newRule(int ID, RuleActiveRegressionNode node, double[] statistics) {
	ActiveRule r=newRule(ID);

	if (node!=null)
	{
	if(node.getPerceptron()!=null)
	{
	r.getLearningNode().setPerceptron(new Perceptron(node.getPerceptron()));
	r.getLearningNode().getPerceptron().setLearningRatio(this.learningRatio);
	}
	if (statistics==null)
	{
	double mean;
	if(node.getNodeStatistics().getValue(0)>0){
	mean=node.getNodeStatistics().getValue(1)/node.getNodeStatistics().getValue(0);
	r.getLearningNode().getTargetMean().reset(mean, 1);
	}
	}
	}
	if (statistics!=null && ((RuleActiveRegressionNode)r.getLearningNode()).getTargetMean()!=null)
	{
	double mean;
	if(statistics[0]>0){
	mean=statistics[1]/statistics[0];
	((RuleActiveRegressionNode)r.getLearningNode()).getTargetMean().reset(mean, (long)statistics[0]);
	}
	}
	return r;
	}

	private ActiveRule newRule(int ID) {
	ActiveRule r=new ActiveRule.Builder().
	threshold(this.pageHinckleyThreshold).
	alpha(this.pageHinckleyAlpha).
	changeDetection(this.driftDetection).
	predictionFunction(this.predictionFunction).
	statistics(new double[3]).
	learningRatio(this.learningRatio).
	numericObserver(numericObserver).
	id(ID).build();
	return r;
	}

	/*
	* Init processor
	*/
	@Override
	public void onCreate(int id) {
	this.processorId = id;
	this.statistics= new double[]{0.0,0,0};
	this.ruleNumberID=0;
	this.defaultRule = newRule(++this.ruleNumberID);

	this.ruleSet = new LinkedList<ActiveRule>();
	}

	/*
	* Clone processor
	*/
	@Override
	public Processor newProcessor(Processor p) {
	AMRulesRegressorProcessor oldProcessor = (AMRulesRegressorProcessor) p;
	Builder builder = new Builder(oldProcessor);
	AMRulesRegressorProcessor newProcessor = builder.build();
	newProcessor.resultStream = oldProcessor.resultStream;
	return newProcessor;
	}

	/*
	* Output stream
	*/
	public void setResultStream(Stream resultStream) {
	this.resultStream = resultStream;
	}

	public Stream getResultStream() {
	return this.resultStream;
	}

	/*
	* Others
	*/
	public boolean isRandomizable() {
	return true;
	}

	/*
	* Builder
	*/
	public static class Builder {
	private int pageHinckleyThreshold;
	private double pageHinckleyAlpha;
	private boolean driftDetection;
	private int predictionFunction; // Adaptive=0 Perceptron=1 TargetMean=2
	private boolean constantLearningRatioDecay;
	private double learningRatio;
	private double splitConfidence;
	private double tieThreshold;
	private int gracePeriod;

	private boolean noAnomalyDetection;
	private double multivariateAnomalyProbabilityThreshold;
	private double univariateAnomalyprobabilityThreshold;
	private int anomalyNumInstThreshold;

	private boolean unorderedRules;

	private FIMTDDNumericAttributeClassLimitObserver numericObserver;
	private ErrorWeightedVote voteType;

	private Instances dataset;

	public Builder(Instances dataset){
	this.dataset = dataset;
	}

	public Builder(AMRulesRegressorProcessor processor) {
	this.pageHinckleyThreshold = processor.pageHinckleyThreshold;
	this.pageHinckleyAlpha = processor.pageHinckleyAlpha;
	this.driftDetection = processor.driftDetection;
	this.predictionFunction = processor.predictionFunction;
	this.constantLearningRatioDecay = processor.constantLearningRatioDecay;
	this.learningRatio = processor.learningRatio;
	this.splitConfidence = processor.splitConfidence;
	this.tieThreshold = processor.tieThreshold;
	this.gracePeriod = processor.gracePeriod;

	this.noAnomalyDetection = processor.noAnomalyDetection;
	this.multivariateAnomalyProbabilityThreshold = processor.multivariateAnomalyProbabilityThreshold;
	this.univariateAnomalyprobabilityThreshold = processor.univariateAnomalyprobabilityThreshold;
	this.anomalyNumInstThreshold = processor.anomalyNumInstThreshold;
	this.unorderedRules = processor.unorderedRules;

	this.numericObserver = processor.numericObserver;
	this.voteType = processor.voteType;
	}

	public Builder threshold(int threshold) {
	this.pageHinckleyThreshold = threshold;
	return this;
	}

	public Builder alpha(double alpha) {
	this.pageHinckleyAlpha = alpha;
	return this;
	}

	public Builder changeDetection(boolean changeDetection) {
	this.driftDetection = changeDetection;
	return this;
	}

	public Builder predictionFunction(int predictionFunction) {
	this.predictionFunction = predictionFunction;
	return this;
	}

	public Builder constantLearningRatioDecay(boolean constantDecay) {
	this.constantLearningRatioDecay = constantDecay;
	return this;
	}

	public Builder learningRatio(double learningRatio) {
	this.learningRatio = learningRatio;
	return this;
	}

	public Builder splitConfidence(double splitConfidence) {
	this.splitConfidence = splitConfidence;
	return this;
	}

	public Builder tieThreshold(double tieThreshold) {
	this.tieThreshold = tieThreshold;
	return this;
	}

	public Builder gracePeriod(int gracePeriod) {
	this.gracePeriod = gracePeriod;
	return this;
	}

	public Builder noAnomalyDetection(boolean noAnomalyDetection) {
	this.noAnomalyDetection = noAnomalyDetection;
	return this;
	}

	public Builder multivariateAnomalyProbabilityThreshold(double mAnomalyThreshold) {
	this.multivariateAnomalyProbabilityThreshold = mAnomalyThreshold;
	return this;
	}

	public Builder univariateAnomalyProbabilityThreshold(double uAnomalyThreshold) {
	this.univariateAnomalyprobabilityThreshold = uAnomalyThreshold;
	return this;
	}

	public Builder anomalyNumberOfInstancesThreshold(int anomalyNumInstThreshold) {
	this.anomalyNumInstThreshold = anomalyNumInstThreshold;
	return this;
	}

	public Builder unorderedRules(boolean unorderedRules) {
	this.unorderedRules = unorderedRules;
	return this;
	}

	public Builder numericObserver(FIMTDDNumericAttributeClassLimitObserver numericObserver) {
	this.numericObserver = numericObserver;
	return this;
	}

	public Builder voteType(ErrorWeightedVote voteType) {
	this.voteType = voteType;
	return this;
	}

	public AMRulesRegressorProcessor build() {
	return new AMRulesRegressorProcessor(this);
	}
	}
	}