<s> import copy <EOL> import numpy <EOL> class StringClassifier ( object ) : <EOL> """<STR_LIT>""" <EOL> _DEFAULT_ALPHA = <NUM_LIT:0.1> <EOL> _DEFAULT_BETA = <NUM_LIT> <EOL> _DEFAULT_TRAINING_ITERATIONS = <NUM_LIT> <EOL> _DEFAULT_PREDICTION_ITERATIONS = <NUM_LIT:5> <EOL> _DEFAULT_PREDICTION_THRESHOLD = <NUM_LIT:0.5> <EOL> """<STR_LIT>""" <EOL> _DEFAULT_MIN_DOCS_TO_PREDICT = <NUM_LIT:20> <EOL> """<STR_LIT>""" <EOL> _DEFAULT_MIN_LABELS_TO_PREDICT = <NUM_LIT:3> <EOL> DEFAULT_LABEL = '<STR_LIT>' <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> numpy . random . seed ( seed = <NUM_LIT:4> ) <EOL> self . _alpha = self . _DEFAULT_ALPHA <EOL> self . _beta = self . _DEFAULT_BETA <EOL> self . _b_dl = None <EOL> self . _c_dl = None <EOL> self . _c_l = None <EOL> self . _c_lw = None <EOL> self . _l_dp = None <EOL> self . _w_dp = None <EOL> self . _label_to_id = None <EOL> self . _word_to_id = None <EOL> self . _num_docs = None <EOL> self . _num_labels = None <EOL> self . _num_words = None <EOL> self . _training_iterations = self . _DEFAULT_TRAINING_ITERATIONS <EOL> self . _prediction_iterations = self . _DEFAULT_PREDICTION_ITERATIONS <EOL> self . _prediction_threshold = self . _DEFAULT_PREDICTION_THRESHOLD <EOL> def _get_word_id ( self , word ) : <EOL> """<STR_LIT>""" <EOL> if word not in self . _word_to_id : <EOL> self . _word_to_id [ word ] = self . _num_words <EOL> self . _num_words += <NUM_LIT:1> <EOL> return self . _word_to_id [ word ] <EOL> def _get_label_id ( self , label ) : <EOL> """<STR_LIT>""" <EOL> if label not in self . _label_to_id : <EOL> self . _label_to_id [ label ] = self . _num_labels <EOL> self . _num_labels += <NUM_LIT:1> <EOL> return self . _label_to_id [ label ] <EOL> def _get_label_name ( self , l ) : <EOL> """<STR_LIT>""" <EOL> for label_name , label_id in self . _label_to_id . iteritems ( ) : <EOL> if label_id == l : <EOL> return label_name <EOL> raise Exception ( '<STR_LIT>' % l ) <EOL> def _get_doc_with_label_vector ( self , d ) : <EOL> """<STR_LIT>""" <EOL> return self . _w_dp [ d ] , self . _b_dl [ d ] <EOL> def _get_label_vector ( self , labels ) : <EOL> """<STR_LIT>""" <EOL> label_vector = numpy . zeros ( self . _num_labels ) <EOL> for label in labels : <EOL> label_vector [ self . _get_label_id ( label ) ] = <NUM_LIT:1> <EOL> label_vector [ self . _label_to_id [ self . DEFAULT_LABEL ] ] = <NUM_LIT:1> <EOL> return label_vector <EOL> def _update_counting_matrices ( self , d , w , l , val ) : <EOL> """<STR_LIT>""" <EOL> self . _c_dl [ d , l ] += val <EOL> self . _c_lw [ l , w ] += val <EOL> self . _c_l [ l ] += val <EOL> def _increment_counting_matrices ( self , d , w , l ) : <EOL> """<STR_LIT>""" <EOL> self . _update_counting_matrices ( d , w , l , <NUM_LIT:1> ) <EOL> def _decrement_counting_matrices ( self , d , w , l ) : <EOL> """<STR_LIT>""" <EOL> self . _update_counting_matrices ( d , w , l , - <NUM_LIT:1> ) <EOL> def _run_gibbs_sampling ( self , doc_ids ) : <EOL> """<STR_LIT>""" <EOL> if doc_ids is None : <EOL> doc_ids = xrange ( self . _num_docs ) <EOL> statez = { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> for d in doc_ids : <EOL> doc , labels = self . _get_doc_with_label_vector ( d ) <EOL> for p , w in enumerate ( doc ) : <EOL> l = self . _l_dp [ d ] [ p ] <EOL> self . _decrement_counting_matrices ( d , w , l ) <EOL> coeff_a = <NUM_LIT:1.0> / ( <EOL> self . _c_dl [ d ] . sum ( ) + self . _num_labels * self . _alpha ) <EOL> coeff_b = <NUM_LIT:1.0> / ( <EOL> self . _c_lw . sum ( axis = <NUM_LIT:1> ) + self . _num_words * self . _beta ) <EOL> label_probabilities = ( <EOL> labels * <EOL> coeff_a * ( self . _c_dl [ d ] + self . _alpha ) * <EOL> coeff_b * ( self . _c_lw [ : , w ] + self . _beta ) ) <EOL> new_label = numpy . random . multinomial ( <EOL> <NUM_LIT:1> , <EOL> label_probabilities / label_probabilities . sum ( ) ) . argmax ( ) <EOL> statez [ '<STR_LIT>' ] += <NUM_LIT:1> <EOL> if l != new_label : <EOL> statez [ '<STR_LIT>' ] += <NUM_LIT:1> <EOL> self . _l_dp [ d ] [ p ] = new_label <EOL> self . _increment_counting_matrices ( d , w , new_label ) <EOL> return statez <EOL> def _get_label_probabilities ( self , d ) : <EOL> """<STR_LIT>""" <EOL> unnormalized_label_probs = ( <EOL> self . _c_dl [ d ] + ( self . _b_dl [ d ] * self . _alpha ) ) <EOL> label_probabilities = ( <EOL> unnormalized_label_probs / unnormalized_label_probs . sum ( ) ) <EOL> return label_probabilities <EOL> def _get_prediction_report_for_doc ( self , d ) : <EOL> """<STR_LIT>""" <EOL> default_label_id = self . _get_label_id ( self . DEFAULT_LABEL ) <EOL> prediction_label_id = default_label_id <EOL> prediction_confidence = <NUM_LIT:0> <EOL> label_probabilities = self . _get_label_probabilities ( d ) <EOL> normalization_coeff = ( <EOL> <NUM_LIT:1.0> / ( <NUM_LIT:1.0> - label_probabilities [ default_label_id ] ) ) <EOL> for l , prob in enumerate ( label_probabilities ) : <EOL> if ( l != default_label_id and <EOL> prob * normalization_coeff > self . _prediction_threshold and <EOL> prob * normalization_coeff > prediction_confidence ) : <EOL> prediction_label_id = l <EOL> prediction_confidence = prob * normalization_coeff <EOL> return { <EOL> '<STR_LIT>' : prediction_label_id , <EOL> '<STR_LIT>' : <EOL> self . _get_label_name ( prediction_label_id ) , <EOL> '<STR_LIT>' : prediction_confidence , <EOL> '<STR_LIT>' : label_probabilities <EOL> } <EOL> def _parse_examples ( self , examples ) : <EOL> """<STR_LIT>""" <EOL> docs = [ ] <EOL> labels_list = [ ] <EOL> for example in examples : <EOL> doc_string = example [ <NUM_LIT:0> ] <EOL> doc = doc_string . split ( ) <EOL> if len ( doc ) > <NUM_LIT:0> : <EOL> labels = example [ <NUM_LIT:1> ] <EOL> docs . append ( doc ) <EOL> labels_list . append ( labels ) <EOL> return docs , labels_list <EOL> def _iterate_gibbs_sampling ( self , iterations , doc_ids ) : <EOL> """<STR_LIT>""" <EOL> for _ in xrange ( iterations ) : <EOL> self . _run_gibbs_sampling ( doc_ids ) <EOL> def _add_examples ( self , examples , iterations ) : <EOL> """<STR_LIT>""" <EOL> if len ( examples ) == <NUM_LIT:0> : <EOL> return <EOL> docs , labels_list = self . _parse_examples ( examples ) <EOL> last_num_labels = self . _num_labels <EOL> last_num_docs = self . _num_docs <EOL> last_num_words = self . _num_words <EOL> for labels in labels_list : <EOL> for label in labels : <EOL> self . _get_label_id ( label ) <EOL> self . _num_docs += len ( docs ) <EOL> self . _b_dl = numpy . concatenate ( <EOL> ( self . _b_dl , numpy . zeros ( <EOL> ( last_num_docs , self . _num_labels - last_num_labels ) , <EOL> dtype = int ) ) , axis = <NUM_LIT:1> ) <EOL> self . _b_dl = numpy . concatenate ( <EOL> ( self . _b_dl , [ <EOL> self . _get_label_vector ( labels ) for labels in labels_list <EOL> ] ) , axis = <NUM_LIT:0> ) <EOL> self . _w_dp . extend ( [ map ( self . _get_word_id , doc ) for doc in docs ] ) <EOL> self . _c_dl = numpy . concatenate ( <EOL> ( self . _c_dl , numpy . zeros ( <EOL> ( last_num_docs , self . _num_labels - last_num_labels ) , <EOL> dtype = int ) ) , axis = <NUM_LIT:1> ) <EOL> self . _c_dl = numpy . concatenate ( <EOL> ( self . _c_dl , numpy . zeros ( <EOL> ( self . _num_docs - last_num_docs , self . _num_labels ) , <EOL> dtype = int ) ) , axis = <NUM_LIT:0> ) <EOL> self . _c_lw = numpy . concatenate ( <EOL> ( self . _c_lw , numpy . zeros ( <EOL> ( last_num_labels , self . _num_words - last_num_words ) , <EOL> dtype = int ) ) , axis = <NUM_LIT:1> ) <EOL> self . _c_lw = numpy . concatenate ( <EOL> ( self . _c_lw , numpy . zeros ( <EOL> ( self . _num_labels - last_num_labels , self . _num_words ) , <EOL> dtype = int ) ) , axis = <NUM_LIT:0> ) <EOL> self . _c_l = numpy . concatenate ( <EOL> ( self . _c_l , numpy . zeros ( <EOL> self . _num_labels - last_num_labels , dtype = int ) ) ) <EOL> for d in xrange ( last_num_docs , self . _num_docs ) : <EOL> doc , _ = self . _get_doc_with_label_vector ( d ) <EOL> l_p = numpy . random . random_integers ( <EOL> <NUM_LIT:0> , self . _num_labels - <NUM_LIT:1> , size = len ( doc ) ) . tolist ( ) <EOL> self . _l_dp . append ( l_p ) <EOL> for w , l in zip ( doc , l_p ) : <EOL> self . _increment_counting_matrices ( d , w , l ) <EOL> self . _iterate_gibbs_sampling ( iterations , [ d ] ) <EOL> return xrange ( last_num_docs , self . _num_docs ) <EOL> def add_examples_for_training ( self , training_examples ) : <EOL> """<STR_LIT>""" <EOL> return self . _add_examples ( training_examples , self . _training_iterations ) <EOL> def add_examples_for_predicting ( self , prediction_examples ) : <EOL> """<STR_LIT>""" <EOL> all_labels = self . _label_to_id . keys ( ) <EOL> return self . _add_examples ( <EOL> zip ( prediction_examples , [ <EOL> copy . deepcopy ( all_labels ) for _ in prediction_examples ] ) , <EOL> self . _prediction_iterations ) <EOL> def load_examples ( self , examples ) : <EOL> """<STR_LIT>""" <EOL> docs , labels_list = self . _parse_examples ( examples ) <EOL> label_set = set ( <EOL> [ self . DEFAULT_LABEL ] + <EOL> [ label for labels in labels_list for label in labels ] ) <EOL> self . _num_labels = len ( label_set ) <EOL> self . _label_to_id = dict ( zip ( label_set , xrange ( self . _num_labels ) ) ) <EOL> self . _num_words = <NUM_LIT:0> <EOL> self . _word_to_id = { } <EOL> self . _num_docs = len ( docs ) <EOL> self . _b_dl = numpy . array ( <EOL> map ( self . _get_label_vector , labels_list ) , dtype = int ) <EOL> self . _w_dp = [ map ( self . _get_word_id , doc ) for doc in docs ] <EOL> self . _c_dl = numpy . zeros ( <EOL> ( self . _num_docs , self . _num_labels ) , dtype = int ) <EOL> self . _c_lw = numpy . zeros ( <EOL> ( self . _num_labels , self . _num_words ) , dtype = int ) <EOL> self . _c_l = numpy . zeros ( self . _num_labels , dtype = int ) <EOL> self . _l_dp = [ ] <EOL> for d in xrange ( self . _num_docs ) : <EOL> doc , _ = self . _get_doc_with_label_vector ( d ) <EOL> l_p = numpy . random . random_integers ( <EOL> <NUM_LIT:0> , self . _num_labels - <NUM_LIT:1> , size = len ( doc ) ) . tolist ( ) <EOL> self . _l_dp . append ( l_p ) <EOL> for w , l in zip ( doc , l_p ) : <EOL> self . _increment_counting_matrices ( d , w , l ) <EOL> self . _iterate_gibbs_sampling ( <EOL> self . _training_iterations , <EOL> xrange ( self . _num_docs ) ) <EOL> def predict_label_for_doc ( self , d ) : <EOL> """<STR_LIT>""" <EOL> if ( self . _num_docs < self . _DEFAULT_MIN_DOCS_TO_PREDICT or <EOL> self . _num_labels < self . _DEFAULT_MIN_LABELS_TO_PREDICT ) : <EOL> return self . DEFAULT_LABEL <EOL> return self . _get_prediction_report_for_doc ( d ) [ '<STR_LIT>' ] <EOL> def to_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> model = { } <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _alpha ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _beta ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( <EOL> self . _prediction_threshold ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( <EOL> self . _training_iterations ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( <EOL> self . _prediction_iterations ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _num_labels ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _num_docs ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _num_words ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _label_to_id ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _word_to_id ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _w_dp ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _b_dl ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _l_dp ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _c_dl ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _c_lw ) <EOL> model [ '<STR_LIT>' ] = copy . deepcopy ( self . _c_l ) <EOL> return model <EOL> def from_dict ( self , model ) : <EOL> """<STR_LIT>""" <EOL> self . _alpha = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _beta = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _prediction_threshold = copy . deepcopy ( <EOL> model [ '<STR_LIT>' ] ) <EOL> self . _training_iterations = copy . deepcopy ( <EOL> model [ '<STR_LIT>' ] ) <EOL> self . _prediction_iterations = copy . deepcopy ( <EOL> model [ '<STR_LIT>' ] ) <EOL> self . _num_labels = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _num_docs = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _num_words = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _label_to_id = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _word_to_id = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _w_dp = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _b_dl = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _l_dp = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _c_dl = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _c_lw = copy . deepcopy ( model [ '<STR_LIT>' ] ) <EOL> self . _c_l = copy . deepcopy ( model [ '<STR_LIT>' ] ) </s>
