Data Processing and Transformation Questions

import json, asyncio
from pyarrow import Table, schema, parquet
from pyarrow import csv as pa_csv

# simplified streaming parser using incremental JSON decoder
def stream_events(stream):
    decoder = json.JSONDecoder()
    buffer = ''
    for chunk in stream:
        buffer += chunk
        while True:
            try:
                obj, idx = decoder.raw_decode(buffer)
            except ValueError:
                break
            yield obj
            buffer = buffer[idx:].lstrip()

async def producer(stream, queue):
    for event in stream_events(stream):
        await queue.put(event)
    await queue.put(None)

async def consumer(queue, parquet_writer):
    while True:
        event = await queue.get()
        if event is None:
            break
        event_id = event.get('id')
        header = extract_header(event)            # small
        parquet_writer.write_event(header)
        for item in stream_items(event):          # stream_items yields items without building list
            row = normalize_item(event_id, item)
            parquet_writer.write_item(row)

from postal.parser import parse_address  # libpostal
import re
import requests

def regex_preclean(s):
    s = s.replace('\n', ', ')
    s = re.sub(r'\s+', ' ', s).strip()
    s = re.sub(r'\bNo\.?\b', 'No', s)
    return s

def parse_raw(raw):
    s = regex_preclean(raw)
    parts = dict(parse_address(s))  # e.g. {'house_number':'10', 'road':'Downing St', 'city':'London',...}
    out = {
        'street_address': ' '.join(filter(None, [parts.get('house_number'), parts.get('road'), parts.get('suburb')])),
        'city': parts.get('city') or parts.get('suburb'),
        'state': parts.get('state'),
        'postal_code': parts.get('postcode'),
        'country': parts.get('country')
    }
    # simple confidence scoring
    score = 0
    score += 1 if out['street_address'] else 0
    score += 1 if out['city'] else 0
    score += 1 if out['postal_code'] else 0
    score += 1 if out['country'] else 0
    out['confidence'] = score/4
    return out

def geocode_validate(address_str):
    # use Nominatim for example; in prod use paid geocoder with quotas
    resp = requests.get('https://nominatim.openstreetmap.org/search',
                        params={'q': address_str, 'format': 'jsonv2', 'addressdetails':1},
                        headers={'User-Agent':'addr-parser/1.0'}).json()
    return resp[0] if resp else None

import sqlite3
from collections import deque, defaultdict
from datetime import datetime, timedelta
import json

# SQLite table: user -> JSON{window:[(date,str,value)], sum:int, last_seen:date}
conn = sqlite3.connect('state.db')
conn.execute('CREATE TABLE IF NOT EXISTS state (user TEXT PRIMARY KEY, payload TEXT)')
conn.commit()

WINDOW_DAYS = 7

def load_state(user):
    row = conn.execute('SELECT payload FROM state WHERE user=?', (user,)).fetchone()
    if not row: return deque(), 0
    p = json.loads(row[0])
    return deque((tuple(x) for x in p['window'])), p['sum']

def save_state(user, window, s):
    payload = json.dumps({'window': list(window), 'sum': s, 'last_seen': window[-1][0]})
    conn.execute('REPLACE INTO state(user,payload) VALUES(?,?)', (user, payload))

def evict_old(window, s, current_date):
    cutoff = (datetime.fromisoformat(current_date) - timedelta(days=WINDOW_DAYS-1)).date()
    while window and datetime.fromisoformat(window[0][0]).date() < cutoff:
        s -= window[0][1]
        window.popleft()
    return s

def process_stream(stream):  # stream yields (user, ts_iso, value)
    out = defaultdict(list)  # date -> list of (user, sum)
    mem_cache = {}  # user -> (window deque, sum)
    FLUSH_INTERVAL = 10000
    i = 0
    for user, ts, val in stream:
        i += 1
        date = ts.split('T')[0]
        if user in mem_cache:
            window, s = mem_cache[user]
        else:
            window, s = load_state(user)
        # append
        window.append((date, val))
        s += val
        s = evict_old(window, s, date)
        mem_cache[user] = (window, s)
        # emit 7-day sum for this date (assuming we want a single sum per user-date)
        out[date].append((user, s))
        # periodic flush to SQLite to bound memory
        if i % FLUSH_INTERVAL == 0:
            for u, (w, ss) in mem_cache.items():
                save_state(u, w, ss)
            conn.commit()
            mem_cache.clear()
    # final flush
    for u, (w, ss) in mem_cache.items():
        save_state(u, w, ss)
    conn.commit()
    return out

WITH base AS (
  SELECT
    transaction_id,
    user_id,
    amount,
    occurred_at
  FROM transactions
),
-- Duplicate detection: count same-amount events within +/-60 seconds for same user
dupes AS (
  SELECT
    t.*,
    -- count of same-amount transactions for this user within 60 seconds (including self)
    COUNT(*) OVER (
      PARTITION BY user_id, amount
      ORDER BY occurred_at
      RANGE BETWEEN INTERVAL '60 seconds' PRECEDING AND INTERVAL '60 seconds' FOLLOWING
    ) AS same_amount_window_count
  FROM base t
),
-- Rolling stats over prior 365 days (exclude current row)
stats AS (
  SELECT
    d.*,
    AVG(amount) OVER (
      PARTITION BY user_id
      ORDER BY occurred_at
      RANGE BETWEEN INTERVAL '365 days' PRECEDING AND INTERVAL '1 second' PRECEDING
    ) AS prior_365_mean,
    STDDEV_SAMP(amount) OVER (
      PARTITION BY user_id
      ORDER BY occurred_at
      RANGE BETWEEN INTERVAL '365 days' PRECEDING AND INTERVAL '1 second' PRECEDING
    ) AS prior_365_stddev,
    COUNT(*) OVER (
      PARTITION BY user_id
      ORDER BY occurred_at
      RANGE BETWEEN INTERVAL '365 days' PRECEDING AND INTERVAL '1 second' PRECEDING
    ) AS prior_365_count
  FROM dupes d
)
SELECT
  transaction_id,
  user_id,
  amount,
  occurred_at,
  -- duplicate flag: more than 1 same-amount event in the +/-60s window
  CASE WHEN same_amount_window_count > 1 THEN TRUE ELSE FALSE END AS duplicate_flag,
  -- outlier flag: only if we have sufficient prior history (example threshold: >= 30)
  CASE
    WHEN prior_365_count >= 30 AND prior_365_stddev IS NOT NULL
         AND amount > prior_365_mean + 3 * prior_365_stddev
    THEN TRUE
    WHEN prior_365_count < 30 THEN NULL -- insufficient history
    ELSE FALSE
  END AS outlier_flag,
  prior_365_count,
  prior_365_mean,
  prior_365_stddev
FROM stats
ORDER BY user_id, occurred_at;

WITH base AS (
  SELECT transaction_id, user_id, amount, occurred_at,
         UNIX_SECONDS(occurred_at) AS tsec
  FROM `project.dataset.transactions`
),
dupes AS (
  SELECT
    *,
    COUNT(*) OVER (
      PARTITION BY user_id, amount
      ORDER BY tsec
      RANGE BETWEEN 60 PRECEDING AND 60 FOLLOWING
    ) AS same_amount_window_count
  FROM base
),
stats AS (
  SELECT
    *,
    AVG(amount) OVER (
      PARTITION BY user_id
      ORDER BY tsec
      RANGE BETWEEN 31536000 PRECEDING AND 1 PRECEDING
    ) AS prior_365_mean,
    STDDEV_SAMP(amount) OVER (
      PARTITION BY user_id
      ORDER BY tsec
      RANGE BETWEEN 31536000 PRECEDING AND 1 PRECEDING
    ) AS prior_365_stddev,
    COUNT(*) OVER (
      PARTITION BY user_id
      ORDER BY tsec
      RANGE BETWEEN 31536000 PRECEDING AND 1 PRECEDING
    ) AS prior_365_count
  FROM dupes
)
SELECT
  transaction_id, user_id, amount, occurred_at,
  IF(same_amount_window_count > 1, TRUE, FALSE) AS duplicate_flag,
  CASE
    WHEN prior_365_count >= 30 AND prior_365_stddev IS NOT NULL
         AND amount > prior_365_mean + 3 * prior_365_stddev THEN TRUE
    WHEN prior_365_count < 30 THEN NULL
    ELSE FALSE
  END AS outlier_flag,
  prior_365_count, prior_365_mean, prior_365_stddev
FROM stats
ORDER BY user_id, occurred_at;