<jupyter_start><jupyter_text>Preparing data from:https://www.stratosphereips.org/datasets-iot23<jupyter_code>pd.set_option('display.max_columns', None)
parsed_files_folder = 'dataset/parsed/'
processed_dfs_folder = 'dataset/processed/'
shuffled_dfs_folder = 'dataset/shuffled/'
files = [parsed_files_folder + i for i in os.listdir(parsed_files_folder)]
columns_to_drop = ['ts', 'uid', 'detailed-label', 'id.orig_h', 'id.orig_p', 'id.resp_h', 'id.resp_p', 'local_orig', 'local_resp', 'tunnel_parents', 'history']
numerical_columns = ['duration', 'orig_bytes', 'resp_bytes', 'missed_bytes', 'orig_pkts', 'orig_ip_bytes', 'resp_pkts', 'resp_ip_bytes']
categorical_columns = ['proto', 'service', 'conn_state']
class NumericalCheck:
    def __init__(self, col_name):
        self.col_name = col_name
        self.min = np.nan
        self.max = np.nan
    
    def update(self, val):
        if np.isnan(self.min) or val < self.min:
            self.min = val
        if np.isnan(self.max) or val > self.max:
            self.max = val
            
    def __str__(self):
        return f'<{self.col_name} -> [{self.min}; {self.max}]>'
    
    def __repr__(self):
        return self.__str__()
class CategoricalCheck:
    def __init__(self, col_name, upper_bound=3000):
        self.col_name = col_name
        self.uniques = set()
        self.valid = True
        self.upper_bound = upper_bound
        self.count = 0
        
    def _check_bound(self):
        self.count = len(self.uniques)
        if self.count > self.upper_bound:
            self.valid = False
            self.uniques = set()
        
    def add_points(self, points: set):
        if self.valid:
            self.uniques.update(points)
            self._check_bound()
        else:
            self.count += len(points)
     
    def __str__(self):
        return f'<{self.col_name}: {self.count} || {"valid" if self.valid else "invalid"}>'
    
    def __repr__(self):
        return self.__str__()
categorical_checks = {col: CategoricalCheck(col) for col in categorical_columns}
numerical_checks = {col: NumericalCheck(col) for col in numerical_columns}
def check_df(file):
    df = pd.read_parquet(file)
    df = df.drop(columns_to_drop, axis=1)
    for col in numerical_columns:
        df[col] = df[col].replace('-', np.nan).map(float)
        numerical_checks[col].update(df[col].max())
        numerical_checks[col].update(df[col].min())
    
    for col in categorical_columns:
        df[col] = df[col].str.lower()
        categorical_checks[col].add_points(set(df[col]))
for file in progressbar(files, max_value=len(files), redirect_stdout=True):
    check_df(file)
numerical_checks
with open('normalizers.json', 'w+') as file:
    json.dump({k: v.max for k, v in numerical_checks.items()}, file)
categorical_checks
with open('encoders.json', 'w+') as file:
    json.dump({k: list(v.uniques) for k, v in categorical_checks.items()}, file)
with open('normalizers.json', 'r') as file:
    numerical_checks = json.load(file)
    
with open('encoders.json', 'r') as file:
    categorical_checks = json.load(file)<jupyter_output><empty_output><jupyter_text>- normalize numerical_columns- one-hot-encode categorical_columns- df.label -> df.isMalicious (lambda r: r == 'malicious') <- y<jupyter_code>def encode_df(df):
    for col in categorical_columns:
        for val in categorical_checks[col]:
            if val == '-':
                continue
            df[col] = df[col].str.lower()
            df[f'{col}_{val}'] = (df[col] == val).map(lambda v: 1 if v else 0)
    return df.drop(categorical_columns, axis=1)
def normalize_df(df):
    for col in numerical_columns:
        df[col] = df[col].replace('-', np.nan).map(float)
        df[col] /= numerical_checks[col]
    return df
def process_file(file):
    print(f'processing file {file}')
    df = pd.read_parquet(file)
    df = df.drop(columns_to_drop, axis=1)
    df['label'] = (df['label'].str.lower() == 'malicious').map(lambda v: 1 if v else 0)
    df = encode_df(df)
    df = normalize_df(df)
    df.to_parquet(file.replace('parsed', 'processed'))
    print(f'file {file} finished')
with ThreadPool(4) as pool:
    pool.map(process_file, files)
pd.read_parquet('dataset/processed/part_50')
processed_files = [processed_dfs_folder + file for file in os.listdir(processed_dfs_folder)]
double_check = {file: len(pd.read_parquet(file)) == len(pd.read_parquet(file.replace('processed', 'parsed'))) for file in processed_files}
{k: v for k, v in double_check.items() if not v}<jupyter_output><empty_output><jupyter_text>Sorting files and rows<jupyter_code>batch_size = 10 # files
random.shuffle(processed_files)
len(processed_files)
def count_shuffled_files():
    return len(os.listdir(shuffled_dfs_folder))
for i in progressbar(range(0, len(processed_files), batch_size), max_value=round(len(processed_files) / batch_size), redirect_stdout=True):
    df = pd.DataFrame()
    for file in processed_files[i:i+batch_size]:
        df = df.append(pd.read_parquet(file))
    
    df = df.sample(frac=1).reset_index(drop=True)
    df.to_parquet(f'{shuffled_dfs_folder}part_{count_shuffled_files() + 1}')<jupyter_output>100% (34.2 of 34.2) |####################| Elapsed Time: 0:08:19 Time:  0:08:19
{'fraction_non_alphanumeric': 0.14528770789523993, 'fraction_numerical': 0.007837889504874785, 'mean_word_length': 3.6506666666666665, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 18, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '124866', 'license': 'MIT', 'n_tokens_mistral': 1859, 'n_tokens_neox': 1803, 'n_words': 369, 'path': 'iot23_data_prep.ipynb', 'repo_name': 'LRAbbade/IoT_anomaly_detection', 'size': '5231'}