view release on metacpan or  search on metacpan

lib/Algorithm/Classifier/IsolationForest/App/Command/stream.pm  view on Meta::CPAN

package Algorithm::Classifier::IsolationForest::App::Command::stream;

use strict;
use warnings;
use Algorithm::Classifier::IsolationForest         ();
use Algorithm::Classifier::IsolationForest::Online ();
use Algorithm::Classifier::IsolationForest::App -command;
use File::Slurp  qw(read_file write_file);
use Scalar::Util qw(looks_like_number);

sub opt_spec {
	return (
		[
			'm=s',
			'Online model JSON file path/name.  Created if it does not exist; resumed and updated if it does.',
			{ 'default' => 'oiforest_model.json', 'completion' => 'files' }
		],
		[ 'i=s', 'Input CSV to stream through the model, in row order.', { 'completion' => 'files' } ],
		[ 'o=s', 'Output the scores to this file instead of printing.',  { 'completion' => 'files' } ],
		[ 'w',   'If the file specified via -o exists, over write it.' ],
		[ 'd',   'Include the input data in the output.' ],
		[
			'learn-only',
			'Only learn the input (warm-up); no scores are emitted.  May not be combined with --score-only.'
		],
		[
			'score-only',
			'Only score the input against the model as-is; nothing is learned.  May not be combined with --learn-only.'
		],
		[ 'threshold=f', 'Alternative decision threshold to use for the label column. 0 < $val < 1' ],
		[
			'save!',
			'Save the updated model state back to -m after streaming (default on; --no-save to discard).',
			{ 'default' => 1 }
		],

		# creation knobs, used only when -m does not exist yet
		[ 'n=i',         'Number of isolation trees in the ensemble (new models only).' ],
		[ 'window=i',    'Sliding window size; 0 disables forgetting (new models only).' ],
		[ 'eta=i',       'max_leaf_samples: points a leaf accumulates before splitting (new models only).' ],
		[ 'growth=s',    "Leaf split-requirement growth, 'adaptive' or 'fixed' (new models only)." ],
		[ 'subsample=f', 'Per-tree stream subsampling probability, in (0, 1] (new models only).' ],
		[ 's=i',         'Seed int (new models only).' ],
		[
			'c=f',
			'Contamination. Expected fraction of anomalies, in (0, 0.5]; learns the decision threshold from the window (new models only).'
		],
		[
			't=s@',
			'Feature name tag. Pass once per feature (e.g. -t cpu -t mem -t disk); the count must match the number of CSV columns or the command will die (new models only).'
		],
		[
			'mungers=s',
			'JSON file of Algorithm::ToNumberMunger specs, keyed by feature tag (new models only; requires -t). '
				. 'Munged CSV columns may hold raw values; rows are munged before streaming and the spec is '
				. 'saved with the model, so resumed runs munge identically. Scalar mungers only for CSV input.',
			{ 'completion' => 'files' }
		],
		[
			'prototype=s',
			'JSON prototype file to create the model from (new models only): the variable schema and '
				. 'schema_version/schema_description come from it, and its params supply knob defaults that the '
				. 'creation switches override. May not be combined with -t or --mungers. See PROTOTYPES in the '
				. 'module POD.',
			{ 'completion' => 'files' }
		],
	);
} ## end sub opt_spec

lib/Algorithm/Classifier/IsolationForest/App/Command/stream.pm  view on Meta::CPAN

		$overrides{'subsample'}        = $opt->{'subsample'} if defined $opt->{'subsample'};
		$overrides{'seed'}             = $opt->{'s'}         if defined $opt->{'s'};
		$overrides{'contamination'}    = $opt->{'c'}         if defined $opt->{'c'};

		$oif = eval { Algorithm::Classifier::IsolationForest->new_from_prototype( $proto, %overrides ) };
		die( '--prototype, "' . $opt->{'prototype'} . '", failed to create a model: ' . $@ ) if $@;
		$from_proto = 1;
	} ## end if ( !$oif && defined( $opt->{'prototype'}...))

	# Munger spec for a NEW model (an existing model carries its own; the
	# creation knob is ignored then, like the rest of them).
	my $mungers;
	if ( !$oif && defined( $opt->{'mungers'} ) ) {
		require JSON::PP;
		$mungers = eval { JSON::PP->new->decode( scalar read_file( $opt->{'mungers'} ) ) };
		die( '--mungers, "' . $opt->{'mungers'} . '", did not parse as JSON: ' . $@ ) if $@;
		die( '--mungers, "' . $opt->{'mungers'} . '", must be a JSON object of tag => spec' )
			unless ref $mungers eq 'HASH';
	}

	my $has_mungers
		= $oif
		? ( ref $oif->{mungers} eq 'HASH' && %{ $oif->{mungers} } ? 1 : 0 )
		: ( $mungers                                              ? 1 : 0 );

	# --- read the CSV, exactly like `iforest fit` does -------------------
	my @data;
	my $expected_cols;

	my $line_int = 1;
	foreach my $line ( read_file( $opt->{'i'} ) ) {
		chomp($line);
		next if $line =~ /^\s*$/;

		my @fields = split( /,/, $line, -1 );

		if ( !defined($expected_cols) ) {
			$expected_cols = scalar @fields;
			die( 'Line ' . $line_int . ' of "' . $opt->{'i'} . '" has no columns' )
				if $expected_cols < 1;
		} elsif ( scalar @fields != $expected_cols ) {
			die(      'Line '
					. $line_int . ' of "'
					. $opt->{'i'}
					. '" has '
					. scalar(@fields)
					. ' columns but expected '
					. $expected_cols );
		}

		if ( !$has_mungers ) {
			my $col_int = 1;
			for my $field (@fields) {
				die(      'Line '
						. $line_int . ' of "'
						. $opt->{'i'}
						. '" value for column '
						. $col_int . ',"'
						. $field
						. '", does not appear to be a number' )
					unless looks_like_number($field);
				$col_int++;
			} ## end for my $field (@fields)
		} ## end if ( !$has_mungers )

		push @data, \@fields;

		$line_int++;
	} ## end foreach my $line ( read_file( $opt->{'i'} ) )

	# A prototype-created model already carries its tags; hold them to the
	# same CSV-width check the -t path gets.
	if ($from_proto) {
		my $n_tags     = scalar @{ $oif->feature_names };
		my $n_features = defined($expected_cols) ? $expected_cols : 0;
		die(      'Number of prototype feature_names ('
				. $n_tags
				. ') does not match number of CSV columns ('
				. $n_features
				. ')' )
			unless $n_tags == $n_features;
	} ## end if ($from_proto)

	# --- create the model when not resuming --------------------------------
	if ( !$oif ) {
		if ( defined( $opt->{'t'} ) ) {
			my $n_tags     = scalar @{ $opt->{'t'} };
			my $n_features = defined($expected_cols) ? $expected_cols : 0;
			die( 'Number of feature tags (' . $n_tags . ') does not match number of CSV columns (' . $n_features . ')' )
				unless $n_tags == $n_features;
		}
		$oif = Algorithm::Classifier::IsolationForest::Online->new(
			'n_trees'          => $opt->{'n'},
			'window_size'      => $opt->{'window'},
			'max_leaf_samples' => $opt->{'eta'},
			'growth'           => $opt->{'growth'},
			'subsample'        => $opt->{'subsample'},
			'seed'             => $opt->{'s'},
			'contamination'    => $opt->{'c'},
			'feature_names'    => $opt->{'t'},
			'mungers'          => $mungers,
		);
	} ## end if ( !$oif )

	# Munge the raw rows into numbers, then run the numeric validation
	# that was skipped at read time.  Munged into a separate structure so
	# -d still prints the raw input columns as given.
	my $stream_rows = \@data;
	if ($has_mungers) {
		my $munged = $oif->munge_rows( \@data );
		for my $i ( 0 .. $#$munged ) {
			for my $col ( 0 .. $#{ $munged->[$i] } ) {
				die(      'Line '
						. ( $i + 1 ) . ' of "'
						. $opt->{'i'}
						. '" value for column '
						. ( $col + 1 ) . ',"'
						. ( defined $munged->[$i][$col] ? $munged->[$i][$col] : 'undef' )
						. '", is not a number after munging' )
					unless looks_like_number( $munged->[$i][$col] );
			} ## end for my $col ( 0 .. $#{ $munged->[$i] } )
		} ## end for my $i ( 0 .. $#$munged )
		$stream_rows = $munged;
	} ## end if ($has_mungers)

	# --- stream ------------------------------------------------------------
	my $results_string = '';
	if ( $opt->{'learn_only'} ) {
		$oif->learn($stream_rows);
	} else {
		my $scores;
		if ( $opt->{'score_only'} ) {
			$scores = $oif->score_samples($stream_rows);
		} else {
			$scores = $oif->score_learn($stream_rows);
		}

		my $threshold
			= defined $opt->{'threshold'}      ? $opt->{'threshold'}
			: defined $oif->decision_threshold ? $oif->decision_threshold
			:                                    0.5;

		for my $i ( 0 .. $#$scores ) {
			my $label = $scores->[$i] >= $threshold ? 1 : 0;
			if ( $opt->{'d'} ) {
				$results_string .= join( ',', @{ $data[$i] } ) . ',' . $scores->[$i] . ',' . $label . "\n";
			} else {
				$results_string .= $scores->[$i] . ',' . $label . "\n";
			}
		}
	} ## end else [ if ( $opt->{'learn_only'} ) ]

	# Refresh the contamination threshold against the post-stream window so
	# the saved model's default cutoff tracks the stream.
	if ( !$opt->{'score_only'} && defined $oif->{contamination} && $oif->window_count ) {
		$oif->relearn_threshold;
	}

	if ( $opt->{'save'} && !$opt->{'score_only'} ) {
		$oif->save( $opt->{'m'} );
	}

	if ( length $results_string ) {
		if ( !defined( $opt->{'o'} ) ) {
			print $results_string;
		} else {
			write_file( $opt->{'o'}, { 'atomic' => 1 }, $results_string );
		}
	}

	return 1;
} ## end sub execute

return 1;