Python for Data Analysis Questions

import pandas as pd
import numpy as np
import logging

logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())

def normalize_timestamps(series: pd.Series, formats=None, default_tz='UTC') -> pd.Series:
    """
    Parse mixed timestamp representations into timezone-aware UTC datetimes.
    - series: pd.Series of mixed types (str, int, pd.NaT, datetime)
    - formats: optional list of str formats to try first (e.g. ["%Y-%m-%d", "%d/%m/%Y %H:%M:%S"])
    - default_tz: timezone to assume for naive datetimes before converting to UTC
    Returns: pd.Series of dtype datetime64[ns, UTC], preserving index; unparsable -> NaT
    """
    s = series.copy()
    n = len(s)
    parsed = pd.Series([pd.NaT]*n, index=s.index)

    # 1) Try user-provided formats
    if formats:
        for fmt in formats:
            mask_unparsed = parsed.isna()
            if not mask_unparsed.any():
                break
            try:
                parsed_loc = pd.to_datetime(s[mask_unparsed], format=fmt, errors='coerce')
            except Exception:
                # format may raise for some input types; coerce anyway elementwise as fallback
                parsed_loc = pd.to_datetime(s[mask_unparsed].astype(str), format=fmt, errors='coerce')
            parsed.loc[mask_unparsed] = parsed_loc

    # 2) Try epoch integers (seconds or milliseconds)
    mask_unparsed = parsed.isna()
    if mask_unparsed.any():
        candidates = s[mask_unparsed]
        # treat ints or digit-only strings as potential epoch
        is_int_like = candidates.apply(lambda x: isinstance(x, (int, np.integer)) or (isinstance(x, str) and x.isdigit()))
        epoch_idx = candidates[is_int_like].index
        if len(epoch_idx):
            vals = s.loc[epoch_idx].astype(np.int64)
            # guess seconds vs milliseconds by magnitude
            # if value > 1e12 -> probably nanoseconds or milliseconds; treat >1e12 as ns, >1e11 as ms
            def to_ts(v):
                if v == 0:
                    return pd.NaT
                if abs(v) > 1e14:   # very large: likely ns
                    return pd.to_datetime(v, unit='ns', errors='coerce')
                if abs(v) > 1e12:
                    return pd.to_datetime(v, unit='us', errors='coerce')
                if abs(v) > 1e10:
                    return pd.to_datetime(v, unit='ms', errors='coerce')
                # otherwise seconds
                return pd.to_datetime(v, unit='s', errors='coerce')
            parsed_epoch = vals.apply(to_ts)
            parsed.loc[epoch_idx] = parsed_epoch

    # 3) Fallback to flexible parser
    mask_unparsed = parsed.isna()
    if mask_unparsed.any():
        parsed.loc[mask_unparsed] = pd.to_datetime(s[mask_unparsed], errors='coerce', utc=False)

    # 4) Localize naive tz and convert to UTC
    # If tz-aware, convert to UTC; if naive, localize to default_tz then convert
    def make_aware(ts):
        if pd.isna(ts):
            return pd.NaT
        if ts.tzinfo is None:
            try:
                return ts.tz_localize(default_tz)
            except Exception:
                # pandas Timestamp may not support tz_localize if already tz-aware; fallback
                return ts.tz_localize(default_tz, ambiguous='NaT', nonexistent='shift_forward')
        return ts

    # Ensure we work with dtype datetime64[ns] or Timestamp objects
    parsed = parsed.apply(lambda x: make_aware(x) if isinstance(x, pd.Timestamp) else (pd.NaT if pd.isna(x) else pd.Timestamp(x).tz_localize(default_tz)))
    # Convert all to UTC
    parsed = parsed.apply(lambda x: x.tz_convert('UTC') if isinstance(x, pd.Timestamp) and x.tzinfo is not None else pd.NaT)

    # 5) Log failures
    failures = parsed[parsed.isna()]
    if not failures.empty:
        failed_examples = failures.index.tolist()[:20]
        logger.warning("normalize_timestamps: %d unparsable entries. Example indices: %s", len(failures), failed_examples)

    # Return as Series with dtype datetime64[ns, UTC]
    return pd.to_datetime(parsed)

import pandas as pd
df = pd.DataFrame({'a':[1, None, 3], 'b':[4,5,None]})
df.dropna(axis=0, how='any')   # drops rows with any NA
df.dropna(axis=1, thresh=2)    # drop cols with <2 non-nulls

df['a'].fillna(df['a'].median(), inplace=True)
df.fillna({'a':0,'b':df['b'].mean()})

df = pd.DataFrame({'t':[1,2,4,5],'v':[10,None,14,15]}).set_index('t')
df['v'].interpolate(method='linear')       # linear by index
df['v'].interpolate(method='time')         # if datetime index

df.sort_values('timestamp', inplace=True)
df.groupby('device')['value'].ffill()   # forward-fill per device
df['value'].bfill(limit=1)              # only fill one step backward

from sklearn.impute import KNNImputer
imputer = KNNImputer(n_neighbors=5)
df_imputed = pd.DataFrame(imputer.fit_transform(df), columns=df.columns)

import cProfile, pstats
cProfile.run("my_aggregation(df)", "prof.out")
p = pstats.Stats("prof.out"); p.sort_stats("cumtime").print_stats(30)

from memory_profiler import profile
@profile
def my_aggregation(df): ...