outrank.core_utils

  1from __future__ import annotations
  2
  3import csv
  4import glob
  5import json
  6import logging
  7import os
  8from collections import Counter
  9from collections import defaultdict
 10from dataclasses import dataclass
 11from typing import Any
 12from typing import Dict
 13from typing import List
 14from typing import Optional
 15from typing import Set
 16from typing import Tuple
 17from typing import Union
 18
 19import numpy as np
 20import pandas as pd
 21import xxhash
 22
 23logging.basicConfig(format='%(asctime)s %(message)s', level=logging.INFO)
 24
 25pro_tips = [
 26    'OutRank can construct subfeatures; features based on subspaces. Example command argument is: --subfeature_mapping "feature_a->feature_b;feature_c<->feature_d;feature_c<->feature_e"',
 27    'Heuristic MI-numba-randomized seems like the best of both worlds! (speed + performance).',
 28    'Heuristic surrogate-lr performs cross-validation (internally), keep that in mind!',
 29    'Consider running OutRank on a smaller data sample first, might be enough (--subsampling = a lot).',
 30    'There are two types of combinations supported; unsupervised pairwise ranking (redundancies- --target_ranking_only=False), and supervised combinations - (--interaction_order > 1)',
 31    'Visualization part also includes clustering - this might be very insightful!',
 32    'By default OutRank includes feature cardinality and coverage in feature names (card; cov)',
 33    'Intermediary checkpoints (tmp_checkpoint.tsv) might already give you insights during longer runs.',
 34    'In theory, you can rank redundancies of combined features (--interaction_order AND --target_ranking_only=False).',
 35    'Give it as many threads as physically possible (--num_threads).',
 36    'You can speed up ranking by diminishing feature buffer size (--combination_number_upper_bound determines how many ranking computations per batch will be considered). This, and --subsampling are very powerful together.',
 37    'Want to rank feature transformations, but not sure which ones to choose? --transformers=default should serve as a solid baseline (common DS transformations included).',
 38    'Your target can be any feature! (explaining one feature with others)',
 39    'OutRank uses HyperLogLog for cardinality estimation - this is also a potential usecase (understanding cardinalities across different data sets).',
 40    'Each feature is named as featureName(cardinality, coverage in percents) in the final files.',
 41    'You can generate candidate feature transformation ranges (fw) by using --task=feature_summary_transformers.',
 42]
 43
 44
 45def write_json_dump_to_file(args: Any, config_name: str) -> None:
 46
 47    out_content = json.dumps(args.__dict__)
 48    with open(config_name, 'w') as out_config:
 49        out_config.write(out_content)
 50
 51
 52def internal_hash(input_obj: str) -> str:
 53    """A generic internal hash used throughout ranking procedure - let's hardcode seed here for sure"""
 54    return xxhash.xxh32(input_obj, seed=20141025).hexdigest()
 55
 56
 57@dataclass
 58class DatasetInformationStorage:
 59    """A generic class for holding properties of a given type of dataset"""
 60
 61    data_path: str
 62    column_names: list[str]
 63    column_types: set[str]
 64    col_delimiter: str | None
 65    encoding: str
 66    fw_map: dict[str, str] | None
 67
 68
 69@dataclass
 70class NumericFeatureSummary:
 71    """A generic class storing numeric feature statistics"""
 72
 73    feature_name: str
 74    minimum: float
 75    maximum: float
 76    median: float
 77    num_unique: int
 78
 79
 80@dataclass
 81class NominalFeatureSummary:
 82    """A generic class storing numeric feature statistics"""
 83
 84    feature_name: str
 85    num_unique: int
 86
 87
 88@dataclass
 89class BatchRankingSummary:
 90    """A generic class representing batched ranking results"""
 91
 92    triplet_scores: list[tuple[str, str, float]]
 93    step_times: dict[str, Any]
 94
 95
 96def display_random_tip() -> None:
 97    TIP_CONTENT = np.random.choice(pro_tips)
 98    tip_core = f"""
 99=====>
100Random tip: {TIP_CONTENT}
101=====>
102    """
103
104    print(tip_core)
105
106
107def get_dataset_info(args: Any):
108    if args.data_source == 'ob-raw-dump':
109        dataset_info = parse_ob_raw_feature_information(args.data_path)
110
111    elif args.data_source == 'ob-vw':
112        dataset_info = parse_ob_vw_feature_information(args.data_path)
113
114    elif args.data_source == 'ob-csv':
115        dataset_info = parse_csv_with_description_information(args.data_path)
116
117    elif args.data_source == 'csv-raw':
118        dataset_info = parse_csv_raw(args.data_path)
119    else:
120        raise NotImplementedError(
121            'Plase, select a supported data source. Possible sources: {csv-raw, ob-vw, ob-csv}',
122        )
123
124    return dataset_info
125
126
127def display_tool_name() -> None:
128    tool_name = """
129
130
131                        *///////////////.
132                     //////////////////////*
133                   */////////////////////////.
134                  ////////////// */////////////
135                  /////////*          /////////
136                 //////   /////   ////,   /////
137                  ////////     ///    /////////
138                  /////   /////  ./////   ////*
139                   ,////                 ////
140                     *////             ////.
141                         ///////*///////
142
143
144    ░█████╗░██╗░░░██╗████████╗██████╗░░█████╗░███╗░░██╗██╗░░██╗
145    ██╔══██╗██║░░░██║╚══██╔══╝██╔══██╗██╔══██╗████╗░██║██║░██╔╝
146    ██║░░██║██║░░░██║░░░██║░░░██████╔╝███████║██╔██╗██║█████═╝░
147    ██║░░██║██║░░░██║░░░██║░░░██╔══██╗██╔══██║██║╚████║██╔═██╗░
148    ╚█████╔╝╚██████╔╝░░░██║░░░██║░░██║██║░░██║██║░╚███║██║░╚██╗
149    ░╚════╝░░╚═════╝░░░░╚═╝░░░╚═╝░░╚═╝╚═╝░░╚═╝╚═╝░░╚══╝╚═╝░░╚═╝
150
151
152    """
153
154    print(tool_name)
155
156
157def parse_ob_line(
158    line_string: str, delimiter: str = '\t', args: Any = None,
159) -> list[str]:
160    """Outbrain line parsing - generic TSVs"""
161
162    line_string = line_string.strip()
163    parts = line_string.split(delimiter)
164    return parts
165
166
167def parse_ob_line_vw(
168    line_string: str,
169    delimiter: str,
170    args: Any = None,
171    fw_col_mapping  = None,
172    table_header  = None,
173    include_namespace_info = False,
174) -> list[str | None]:
175    """Parse a sparse vw line into a pandas df with pre-defined namespace"""
176
177    all_line_parts = line_string.strip().split('|')
178    label_part = all_line_parts[0].split(' ')[0]
179    remainder = all_line_parts[1:]
180    label = label_part
181    remainder_hash = dict()
182
183    # Hash multi-value tuples and store name-val mappings
184    for remaining_part in remainder:
185        core_parts = remaining_part.strip().split(' ')
186        namespace_part = core_parts[0]
187        other_parts = '-'.join(x for x in core_parts[1:] if x != '')
188
189        if namespace_part in fw_col_mapping:
190            remainder_hash[fw_col_mapping[namespace_part]] = other_parts
191        else:
192            logging.error(f"Didn't find namespace {namespace_part}")
193
194    # Construct the consistently-mapped instance based on the remainder mapping
195    the_real_instance = [
196        remainder_hash.get(
197            el, None,
198        ) for el in table_header[1:]
199    ]
200    if not include_namespace_info:
201        the_real_instance = [
202            x[2:] if not x is None else None for x in the_real_instance
203        ]
204
205    parts = [label] + the_real_instance
206
207    return parts
208
209
210def parse_ob_csv_line(
211    line_string: str, delimiter: str = ',', args: Any = None,
212) -> list[str]:
213    """Data can have commas within JSON field dumps"""
214
215    clx = list(csv.reader([line_string])).pop()
216    return clx
217
218
219def generic_line_parser(
220    line_string: str,
221    delimiter: str,
222    args: Any = None,
223    fw_col_mapping: Any = None,
224    table_header: Any = None,
225) -> list[Any]:
226    """A generic method aimed to parse data from different sources."""
227
228    if args.data_source == 'ob-raw-dump':
229        return parse_ob_line(line_string, delimiter, args)
230
231    elif args.data_source == 'ob-vw':
232        return parse_ob_line_vw(
233            line_string, delimiter, args, fw_col_mapping, table_header,
234        )
235
236    elif args.data_source == 'ob-csv' or args.data_source == 'csv-raw':
237        return parse_ob_csv_line(line_string, delimiter, args)
238
239    else:
240        raise NotImplementedError(
241            'Please, specify a valid --data_source argument!',
242        )
243
244
245def read_reference_json(json_path) -> dict[str, dict]:
246    """A helper method for reading a JSON"""
247    with open(json_path) as jp:
248        return json.load(jp)
249
250
251def parse_namespace(namespace_path: str) -> tuple[set[str], dict[str, str]]:
252    """Parse the feature namespace for type awareness"""
253
254    float_set = set()
255    id_feature_map = {}
256
257    with open(namespace_path) as nm:
258        for line in nm:
259            try:
260                namespace_parts = line.strip().split(',')
261                if len(namespace_parts) == 2 and '_' not in namespace_parts[0]:
262                    fw_id, feature = namespace_parts
263                    type_name = 'generic'
264
265                else:
266                    fw_id, feature, type_name = namespace_parts
267
268                id_feature_map[fw_id] = feature
269                if type_name == 'f32':
270                    float_set.add(feature)
271            except Exception as es:
272                pass
273
274    return float_set, id_feature_map
275
276
277def read_column_names(mapping_file: str) -> list[str]:
278    """Read the col. header"""
279
280    with open(mapping_file, encoding='utf-8') as mf:
281        columns = mf.read().strip().split('\t')
282    return columns
283
284
285def parse_ob_vw_feature_information(data_path) -> DatasetInformationStorage:
286    """A generic parser of ob-based data"""
287
288    # Get column names
289    column_descriptions = os.path.join(data_path, 'vw_namespace_map.csv')
290    column_types, fw_map = parse_namespace(column_descriptions)
291
292    # We establish column order here
293    column_names = ['label'] + list(fw_map.values())
294
295    data_path = os.path.join(data_path, 'data.vw.gz')
296    col_delimiter = None
297    encoding = 'utf-8'
298
299    return DatasetInformationStorage(
300        data_path, column_names, column_types, col_delimiter, encoding, fw_map,
301    )
302
303
304def parse_ob_raw_feature_information(data_path) -> DatasetInformationStorage:
305    """A generic parser of ob-based data"""
306
307    # Get column names
308    column_types: list[str] = []
309
310    # Get set of numeric columns
311    table_header_path = os.path.join(data_path, 'raw_data/0_header/header.csv')
312    table_header = read_column_names(table_header_path)
313
314    data_path_train = os.path.join(data_path, 'raw_data/1_train/*')
315    col_delimiter = '\t'
316    encoding = 'utf-8'
317
318    final_df = []
319    core_data_folders = glob.glob(data_path_train)
320    for actual_data in core_data_folders:
321        for dump in glob.glob(actual_data + '/*'):
322            tmp_df = pd.read_csv(
323                dump, sep='\t', low_memory=True, dtype='object',
324            )
325            assert tmp_df.shape[1] == len(table_header)
326            tmp_df.columns = table_header
327            final_df.append(tmp_df)
328
329    final_df_concat = pd.concat(final_df, axis=0)
330    final_path = os.path.join(data_path, 'raw_dump.tsv')
331    logging.info(
332        f'Stored data dump of dimension {final_df_concat.shape} to {final_path}',
333    )
334    final_df_concat.to_csv(final_path, sep='\t', index=False)
335    data_path = os.path.join(data_path, 'raw_dump.tsv')
336
337    return DatasetInformationStorage(
338        data_path, table_header, set(column_types), col_delimiter, encoding, None,
339    )
340
341
342def parse_ob_feature_information(data_path) -> DatasetInformationStorage:
343    """A generic parser of ob-based data"""
344
345    # Get column names
346    column_names = os.path.join(data_path, 'vw_namespace_map.csv')
347    column_types, _ = parse_namespace(column_names)
348
349    # Get set of numeric columns
350    table_header_path = os.path.join(data_path, 'raw_data/0_header/header.csv')
351    table_header = read_column_names(table_header_path)
352
353    data_path = os.path.join(data_path, 'raw_data/1_train/*')
354    col_delimiter = '\t'
355    encoding = 'utf-8'
356
357    return DatasetInformationStorage(
358        data_path, table_header, column_types, col_delimiter, encoding, None,
359    )
360
361
362def parse_csv_with_description_information(data_path) -> DatasetInformationStorage:
363    dataset_description = read_reference_json(
364        os.path.join(data_path, 'dataset_desc.json'),
365    )
366    column_names = []
367    column_types = set()
368    for feature in dataset_description.get('data_features', []):
369        feature_name = feature.get('name')
370        column_names.append(feature_name)
371        feature_type = feature.get('type', '')
372        if 'float' in feature_type or 'Float' in feature_type:
373            column_types.add(feature_name)
374    col_delimiter = ','
375    data_path = os.path.join(data_path, 'data.csv')
376    encoding = 'latin1'
377    return DatasetInformationStorage(
378        data_path, column_names, column_types, col_delimiter, encoding, None,
379    )
380
381
382def parse_csv_raw(data_path) -> DatasetInformationStorage:
383    column_types: set[str] = set()
384
385    data_path = os.path.join(data_path, 'data.csv')
386    with open(data_path) as inp_data:
387        header = inp_data.readline()
388    col_delimiter = ','
389    column_names = header.strip().split(col_delimiter)
390    encoding = 'latin1'
391    return DatasetInformationStorage(
392        data_path, column_names, column_types, col_delimiter, encoding, None,
393    )
394
395
396def extract_features_from_reference_JSON(json_path: str, combined_features_only = False, all_features = False) -> set[Any]:
397    """Given a model's JSON, extract unique features"""
398
399    with open(json_path) as jp:
400        content = json.load(jp)
401
402    unique_features = set()
403    feature_space = content['desc'].get('features', [])
404    if all_features:
405        return set(feature_space)
406
407    fields_space = content['desc'].get('fields', [])
408    joint_space = feature_space + fields_space
409
410    if combined_features_only:
411        return {feature for feature in feature_space if len(feature.split(','))>1}
412
413    for feature_tuple in joint_space:
414        for individual_feature in feature_tuple.split(','):
415            unique_features.add(individual_feature)
416
417    return unique_features
418
419
420def summarize_feature_bounds_for_transformers(
421    bounds_object_storage: Any,
422    feature_types: list[str],
423    task_name: str,
424    label_name: str,
425    granularity: int = 15,
426    output_summary_table_only: bool = False,
427):
428    """summarization auxilliary method for generating JSON-based specs"""
429
430    if bounds_object_storage is None:
431        logging.info('Bounds storage object is empty.')
432        exit()
433
434    final_storage = defaultdict(list)
435    for el in bounds_object_storage:
436        if isinstance(el, dict):
437            for k, v in el.items():
438                final_storage[k].append(v)
439
440    summary_table_rows = []
441    for k, v in final_storage.items():
442        # Conduct local aggregation + bound changes
443        if k in feature_types and k != label_name:
444            minima, maxima, medians, uniques = [], [], [], []
445            for feature_summary in v:
446                minima.append(feature_summary.minimum)
447                maxima.append(feature_summary.maximum)
448                medians.append(feature_summary.median)
449                uniques.append(feature_summary.num_unique)
450            summary_table_rows.append(
451                [
452                    k,
453                    round(np.min(minima), 2),
454                    round(np.max(maxima), 2),
455                    round(np.median(medians), 2),
456                    int(np.mean(uniques)),
457                ],
458            )
459
460    if len(summary_table_rows) == 0:
461        logging.info('No numeric features to summarize.')
462        return None
463
464    summary_table: pd.Dataframe = pd.DataFrame(summary_table_rows)
465    summary_table.columns = [
466        'Feature',
467        'Minimum',
468        'Maximum',
469        'Median',
470        'Num avg. unique (batch)',
471    ]
472
473    if output_summary_table_only:
474        return summary_table
475
476    if len(summary_table) == 0:
477        logging.info('Summary table empty, skipping transformer generation ..')
478        return
479
480    if task_name == 'feature_summary_transformers':
481        transformers_per_feature = defaultdict(list)
482
483        # Take care of weights first -> range is pre-defined
484        for k, v in final_storage.items():
485            if label_name in k or 'dummy' in k:
486                continue
487
488            weight_template = {
489                'feature': k,
490                'src_features': [k],
491                'transformations': ['Weight'],
492                'weights': [0, 0.5, 1.5, 2, 3, 10],
493            }
494            transformers_per_feature[k].append(weight_template)
495
496        # Consider numeric transformations - pairs and single ones
497        for enx, row in summary_table.iterrows():
498            if row.Feature == 'dummy':
499                continue
500            try:
501                actual_range = (
502                    np.arange(
503                        row['Minimum'],
504                        row['Maximum'],
505                        (row['Maximum'] - row['Minimum']) / granularity,
506                    )
507                    .round(2)
508                    .tolist()
509                )
510                binner_template = {
511                    'feature': f'{row.Feature}',
512                    'src_features': [row.Feature],
513                    'transformations': [
514                        'BinnerSqrt',
515                        'BinnerLog',
516                        'BinnerSqrtPlain',
517                        'BinnerLogPlain',
518                    ],
519                    'n': actual_range,
520                    'resolutions': [0.1, 2, 4, 8, 16, 32, 64, 128],
521                }
522
523            except Exception as es:
524                logging.info(
525                    f'\U0001F631 Encountered {es}. The problematic feature is: {row}, skipping transformer for this feature ..',
526                )
527
528            transformers_per_feature[row.Feature].append(binner_template)
529
530            # We want the full loop here, due to asymmetry of transformation(s)
531            for enx_second, row_second in summary_table.iterrows():
532                if enx_second < enx:
533                    continue
534
535                # The n values are defined based on maxima of the second feature
536                if row_second.Feature != row.Feature:
537                    n_bound = round(row_second['Median'] + row['Median'], 2)
538                    max_bound = round(
539                        min(row_second['Maximum'], row['Maximum']), 2,
540                    )
541                    min_bound = round(
542                        row_second['Minimum'] + row['Minimum'], 2,
543                    )
544                    range_spectrum = sorted(
545                        list(
546                            {
547                                0.0,
548                                min_bound,
549                                n_bound / 10,
550                                n_bound / 5,
551                                n_bound,
552                                max_bound,
553                            },
554                        ),
555                    )
556
557                    range_spectrum = [x for x in range_spectrum if x >= 0]
558                    binner_pair_template = {
559                        'feature': f'{row.Feature}Ratio{row_second.Feature}',
560                        'src_features': [row.Feature, row_second.Feature],
561                        'transformations': ['BinnerLogRatioPlain', 'BinnerLogRatio'],
562                        'n': range_spectrum,
563                        'resolutions': [0.1, 2, 4, 8, 16, 32, 64, 128],
564                    }
565
566                    binner_pair_template_second = {
567                        'feature': f'{row_second.Feature}Ratio{row.Feature}',
568                        'src_features': [row_second.Feature, row.Feature],
569                        'transformations': ['BinnerLogRatioPlain', 'BinnerLogRatio'],
570                        'n': range_spectrum,
571                        'resolutions': [0.1, 2, 4, 8, 16, 32, 64, 128],
572                    }
573
574                    transformers_per_feature[row.Feature].append(
575                        binner_pair_template,
576                    )
577                    transformers_per_feature[row.Feature].append(
578                        binner_pair_template_second,
579                    )
580
581        binner_templates = []
582        for k, v in transformers_per_feature.items():
583            for transformer_struct in v:
584                binner_templates.append(transformer_struct)
585
586        logging.info(
587            f'Generated {len(binner_templates)} transformation search specifications.\n',
588        )
589        namespace_full = f'"random_grid_feature_transform": {json.dumps(binner_templates)}, "random_grid_epochs": 512'
590        logging.info('Generated transformations below:\n')
591        print(namespace_full)
592
593
594def summarize_rare_counts(
595    term_counter: Any,
596    args: Any,
597    cardinality_object: Any,
598    object_info: DatasetInformationStorage,
599) -> None:
600    """Write rare values"""
601
602    out_df_rows = []
603    logging.info(
604        f'Rare value summary (freq <= {args.rare_value_count_upper_bound}) follows ..',
605    )
606
607    for namespace_tuple, count in term_counter.items():
608        namespace, value = namespace_tuple
609        out_df_rows.append([namespace, value, count])
610    out_df: pd.DataFrame = pd.DataFrame(out_df_rows)
611    out_df.columns = ['Namespace', 'value', 'Count']
612    out_df.to_csv(
613        os.path.join(args.output_folder, 'rare_values.tsv'), sep='\t', index=False,
614    )
615    logging.info(f'Wrote rare values to {args.output_folder}/rare_values.tsv')
616
617    overall_rare_counts = Counter(out_df.Namespace.values)
618    sorted_counts = sorted(
619        overall_rare_counts.items(), key=lambda pair: pair[1], reverse=True,
620    )
621    for k, v in sorted_counts:
622        logging.info(f'Namespace: {k} ---- Rare values observed: {v}')
623
624    final_df_rows = []
625    for k, v in sorted_counts:
626        cardinality = len(cardinality_object[k])
627        rare_proportion = np.round(100 * (v / cardinality), 2)
628        col_type = 'nominal'
629        if k in object_info.column_types:
630            col_type = 'numeric'
631        final_df_rows.append(
632            {
633                'rare_proportion': rare_proportion,
634                'feature_type': col_type,
635                'feature_name': k,
636            },
637        )
638
639    final_df: pd.DataFrame = pd.DataFrame(final_df_rows)
640    final_df = final_df.sort_values(by=['rare_proportion'])
641    logging.info(
642        f'Wrote feature sparsity summary to {args.output_folder}/feature_sparsity_summary.tsv',
643    )
644    final_df.to_csv(
645        f'{args.output_folder}/feature_sparsity_summary.tsv', index=False, sep='\t',
646    )
647
648
649def is_prior_heuristic(args: Any) -> bool:
650    if '-prior' in args.heuristic and args.reference_model_JSON:
651        return True
652    return False
653
654
655def get_num_of_instances(fname: str) -> int:
656    """Count the number of lines in a file, fast - useful for progress logging"""
657
658    def _make_gen(reader):
659        while True:
660            b = reader(2**16)
661            if not b:
662                break
663            yield b
664
665    with open(fname, 'rb') as f:
666        count = sum(buf.count(b'\n') for buf in _make_gen(f.raw.read))
667    return count