Pandas Data Manipulation and Analysis Questions

import pandas as pd
import time
import sqlite3
from datasketch import HyperLogLog  # approximate uniques

# in-memory state: DataFrame indexed by user_id
state = pd.DataFrame(columns=['count','hll','last_seen','processed_max_event_ts'])
state = state.set_index(pd.Index([], name='user_id'))

# simple WAL in sqlite for processed event_ids / checkpoints
conn = sqlite3.connect('stream_state.db')
conn.execute('CREATE TABLE IF NOT EXISTS processed(event_id TEXT PRIMARY KEY, user_id TEXT, ts REAL)')

def load_state():
    # load persisted CSV/parquet of state, and reconstruct HLL objects
    df = pd.read_parquet('state.parquet')  # contains serialized hll bytes
    df['hll'] = df['hll'].apply(deserialize_hll)
    return df.set_index('user_id')

def persist_state(df):
    df_to_store = df.copy()
    df_to_store['hll'] = df_to_store['hll'].apply(serialize_hll)
    df_to_store.reset_index().to_parquet('state.parquet', overwrite=True)

def process_batch(events_df):
    global state, conn
    # events_df columns: event_id, user_id, ts, value
    for _, ev in events_df.sort_values('ts').iterrows():
        # idempotency: skip if event_id already in processed WAL
        if conn.execute('SELECT 1 FROM processed WHERE event_id=?',(ev.event_id,)).fetchone():
            continue
        u = ev.user_id
        if u not in state.index:
            state.loc[u] = [0, HyperLogLog(p=12), 0.0, 0.0]
        # update features
        state.at[u,'count'] += 1
        state.at[u,'hll'].update(ev.value.encode('utf8'))
        state.at[u,'last_seen'] = ev.ts
        state.at[u,'processed_max_event_ts'] = max(state.at[u,'processed_max_event_ts'], ev.ts)
        # WAL insert
        conn.execute('INSERT INTO processed(event_id,user_id,ts) VALUES (?,?,?)',(ev.event_id,u,ev.ts))
    conn.commit()
    persist_state(state)

import pandas as pd
import numpy as np

# sample dataframe
df = pd.DataFrame({
    "a": np.random.rand(1_000_000),
    "b": np.random.rand(1_000_000)
})

# compute score and filter using eval + query (both accept expression strings)
df = df.copy()
pd.eval("df['score'] = 0.3 * df.a + 0.7 * df.b", engine="numexpr")  # computes score in-place
result = df.query("score > 0.5")  # returns filtered dataframe

# alternative: compute and filter in one expression (avoids creating 'score' column)
result2 = df.query("0.3 * a + 0.7 * b > 0.5")  # uses the expression directly

import pandas as pd
df = pd.DataFrame({"A":[10,20,30]}, index=["x","y","z"])

# Get scalar by label (fast)
val = df.at["y","A"]

# Get scalar by position (fast)
val2 = df.iat[1,0]

# Set scalar by label (fast)
df.at["y","A"] = 25

# Set scalar by position (fast)
df.iat[2,0] = 35

sub = df[df["A"]>10]   # may be a view or a copy
sub["B"] = 1          # SettingWithCopyWarning

import pandas as pd
import numpy as np

# df: DataFrame with ['user_id','country','amount','timestamp']

agg_df = (
    df.groupby(['user_id', 'country'], observed=True)
      .agg(
          total_amount = ('amount', 'sum'),
          avg_amount   = ('amount', 'mean'),
          transaction_count = ('amount', 'count'),
          p95_amount   = ('amount', lambda x: x.quantile(0.95))
      )
      .reset_index()  # flatten the hierarchical index to columns
)

# optional: ensure numeric types and handle NaNs
agg_df['avg_amount'] = agg_df['avg_amount'].fillna(0)
agg_df['p95_amount'] = agg_df['p95_amount'].fillna(0)

import pandas as pd

def flatten_columns(df, sep="_", na_rep="", order=None):
    """
    Flatten MultiIndex columns to single strings deterministically and
    disambiguate duplicates by appending suffixes.
    """
    cols = df.columns
    # If not MultiIndex, coerce to Index of tuples for uniform handling
    if not isinstance(cols, pd.MultiIndex):
        tuples = [(c,) for c in cols]
        nlevels = 1
    else:
        tuples = list(cols)
        nlevels = cols.nlevels

    # Optional explicit ordering: tuple of level indices; default left-to-right
    if order is None:
        order = list(range(nlevels))

    # Build base names: join level values in 'order'
    def norm(x):
        if x is None:
            return na_rep
        if isinstance(x, str):
            return x
        return str(x)

    base_names = []
    for t in tuples:
        parts = [norm(t[i]) if i < len(t) else na_rep for i in order]
        # remove empty parts to avoid trailing separators if desired
        parts = [p for p in parts if p != ""]
        name = sep.join(parts) if parts else na_rep or "col"
        base_names.append(name)

    # Disambiguate duplicates deterministically
    seen = {}
    final = []
    for name in base_names:
        if name not in seen:
            seen[name] = 0
            final.append(name)
        else:
            seen[name] += 1
            candidate = f"{name}{sep}{seen[name]}"
            # ensure candidate not already used (rare)
            while candidate in seen:
                seen[name] += 1
                candidate = f"{name}{sep}{seen[name]}"
            seen[candidate] = 0
            final.append(candidate)

    df = df.copy()
    df.columns = final
    return df

# Example:
# result = pivot_table(...)  # returns df with MultiIndex columns
# flattened = flatten_columns(result, sep="_")