การตั้งค่าโครงการและการนำเข้าข้อมูล

เป้าหมายของโครงงานสรุปยอด

การกำหนดเป้าหมายโครงงานและ KPI

# Project configuration — define goals upfront
CONFIG = {
    'csv_path': 'data/orders_2024.csv',
    'db_url': 'sqlite:///customer_db.sqlite',
    'db_table': 'customers',
    'output_dir': 'output/',
    'report_path': 'output/report.md',
    'analysis_year': 2024,
    'top_n_regions': 5,
    'rolling_window_days': 30
}

print('Project config loaded.')
print('Target KPIs: monthly revenue, cohort retention, top regions by margin')

การโหลดข้อมูลจาก CSV

import pandas as pd

df_orders = pd.read_csv(
    'data/orders_2024.csv',
    dtype={
        'order_id': 'int32',
        'customer_id': 'int32',
        'product_id': 'int32',
        'quantity': 'int16',
        'unit_price': 'float32',
        'region': 'category',
        'category': 'category'
    },
    parse_dates=['order_date'],
    encoding='utf-8'
)
print(f'Orders loaded: {df_orders.shape}')
print(df_orders.dtypes)
print(df_orders.head(3))

การโหลดข้อมูลจากฐานข้อมูล

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///customer_db.sqlite')

df_customers = pd.read_sql_query(
    '''
    SELECT customer_id, customer_name, country,
           acquisition_channel, signup_date
    FROM customers
    ''',
    con=engine,
    dtype={'customer_id': 'int32'},
    parse_dates=['signup_date']
)
print(f'Customers loaded: {df_customers.shape}')
print(df_customers.dtypes)
print(df_customers.head(3))

การรวมแหล่งข้อมูลสองแหล่ง

import pandas as pd

# Left join: keep all orders, enrich with customer data where available
df = pd.merge(
    df_orders,
    df_customers,
    on='customer_id',
    how='left'
)

print(f'Orders: {len(df_orders)}, Customers: {len(df_customers)}')
print(f'Merged: {df.shape}')
unmatched = df['customer_name'].isna().sum()
print(f'Orders with no matching customer: {unmatched} ({unmatched/len(df):.1%})')

รายการตรวจสอบข้อมูลฉบับสมบูรณ์

import pandas as pd

def audit_dataframe(df, name='DataFrame'):
    print(f'=== Audit: {name} ===')
    print(f'Shape: {df.shape}')
    print(f'Date range: {df["order_date"].min()} to {df["order_date"].max()}')
    print(f'Duplicate order_ids: {df["order_id"].duplicated().sum()}')
    print('Missing values:')
    missing = df.isnull().sum()
    print(missing[missing > 0].to_string())
    print('Negative quantities:', (df['quantity'] < 0).sum())
    print('Negative prices:', (df['unit_price'] < 0).sum())
    print()

audit_dataframe(df, 'Merged Orders + Customers')

การแก้ไขความผิดปกติในการแยกวิเคราะห์วันที่

import pandas as pd
from datetime import datetime

# Flag impossible dates
df['date_valid'] = (
    (df['order_date'] >= '2024-01-01') &
    (df['order_date'] <= datetime.now())
)
print('Invalid order dates:', (~df['date_valid']).sum())

# Flag orders before customer signup
df['signup_date'] = pd.to_datetime(df['signup_date'])
df['date_before_signup'] = df['order_date'] < df['signup_date']
print('Orders before customer signup:', df['date_before_signup'].sum())

# Set invalid dates to NaT
df.loc[~df['date_valid'], 'order_date'] = pd.NaT

การจัดการจำนวนติดลบและการคืนสินค้า

import pandas as pd

# Tag returns
df['is_return'] = df['quantity'] < 0

returns = df[df['is_return']]
orders = df[~df['is_return']]

print(f'Normal orders: {len(orders)}')
print(f'Returns/refunds: {len(returns)} ({len(returns)/len(df):.1%})')
print(f'Return rate by category:')
print(
    df.groupby('category')['is_return']
    .mean()
    .sort_values(ascending=False)
    .round(3)
)

การตรวจสอบความครบถ้วนของการเชื่อมข้อมูล

import pandas as pd

order_customers = set(df_orders['customer_id'].unique())
customer_ids = set(df_customers['customer_id'].unique())

orphans = order_customers - customer_ids
print(f'Customer IDs in orders not in customer table: {len(orphans)}')
print(f'Coverage: {len(order_customers & customer_ids) / len(order_customers):.1%} of order customers have records')

# Revenue from unmatched customers
orphan_revenue = df[df['customer_id'].isin(orphans)]['unit_price'].sum()
print(f'Revenue from orphan customers: ${orphan_revenue:,.0f}')

การบันทึกชุดข้อมูลที่ผ่านการตรวจสอบ

import os
import pandas as pd

OS_DIR = 'output/'
os.makedirs(OS_DIR, exist_ok=True)

# Save clean orders (excluding returns and invalid dates)
clean_orders = df[
    (~df['is_return']) &
    df['order_date'].notna()
].copy()

clean_orders.to_parquet(OS_DIR + 'clean_orders.parquet', index=False)
df[df['is_return']].to_parquet(OS_DIR + 'returns.parquet', index=False)

print(f'Clean orders saved: {len(clean_orders):,} rows')
print(f'Returns saved: {len(df[df["is_return"]]):,} rows')
print(f'Files in {OS_DIR}: {os.listdir(OS_DIR)}')

รายงานสรุปการตรวจสอบ

import pandas as pd
from datetime import datetime

def write_audit_summary(df, path):
    lines = [
        '# Data Ingestion Audit',
        f'Generated: {datetime.now().strftime("%Y-%m-%d %H:%M")}',
        '',
        f'- Total records after merge: {len(df):,}',
        f'- Clean orders: {(~df["is_return"]).sum():,}',
        f'- Returns: {df["is_return"].sum():,}',
        f'- Date range: {df["order_date"].min().date()} to {df["order_date"].max().date()}',
        f'- Unique customers: {df["customer_id"].nunique():,}',
        f'- Unique categories: {df["category"].nunique()}',
        f'- Missing unit_price: {df["unit_price"].isna().sum()}'
    ]
    with open(path, 'w') as f:
        f.write('\n'.join(lines))

write_audit_summary(df, 'output/audit.md')
print('Audit summary written.')

ตรวจสอบความเข้าใจอย่างรวดเร็ว

ทบทวนบทเรียน