Part 7 — Postgres Data Model for a Near-Real-Time Market Copilot

(n8n schema, idempotent ingestion, agent-friendly views, and materialized views for fast reporting)

By the time you’ve built ingestion loops, a news summariser, and a Telegram agent, you’ve already learned the hard truth:

Near-real-time analytics isn’t hard because of SQL — it’s hard because your database becomes both the write path and the read path.

In my “Market Copilot”, n8n is continuously inserting into Postgres: ticks every few minutes, daily bars on a slower cadence, news in bursts, and analyst runs on demand. If dashboards (Metabase) and my Telegram bot hit those raw tables directly, I eventually pay the price:

• slow charts caused by scanning append-only time-series tables
• unpredictable latency during “hot” ingestion windows
• accidental expensive joins (especially with news + ticks + runs)
• hard-to-control access patterns for an LLM-powered agent

This post is about the database design that makes the entire system feel fast, safe, and predictable.

My approach is intentionally pragmatic:

1. keep raw tables, append-first and easy to load into
2. Add idempotency constraints and indexes where they matter
3. create agent-friendly views (v_*) as a read-only contract for the Telegram copilot
4. create materialised views (mv_*) to precompute the dashboard primitives
5. refresh materialised views safely from n8n (no blocking reads, no overlapping refresh jobs)

And importantly, I’m keeping everything inside the n8n schema for now. Simple to operate, simple to reason about.

1) Schema overview (what lives where)

Even though I’m using one schema, I still think in layers:

Raw ingestion tables (write-optimised)

• n8n.assets — asset catalogue (symbols, types, metadata)
• n8n.asset_ticks — high-frequency tick prices
• n8n.asset_daily — daily closes
• n8n.asset_news — news items + summaries + quality/confidence
• n8n.analyst_runs — scoring outputs per asset over time

ERD — Raw Tables

Curated read layer (stable query contract for the agent)

• n8n.v_* — plain views that the Telegram bot is allowed to query

Reporting acceleration layer (dashboard primitives)

• n8n.mv_* — materialised views used by Metabase (and optionally the bot later)

The point is not “warehouse purity”. The point is to provide a stable, safe, and fast interface for consumers while ingestion continues uninterrupted.

2) Table DDL (Create Table statements)

Below are clean, explicit CREATE TABLE statements that match your model and make the important constraints/indexes visible in the article. If your existing tables are already created, you can treat this as documentation (and selectively apply only the indexes/constraints you’re missing).

2.1 n8n.assets

CREATE TABLE IF NOT EXISTS n8n.assets (
  asset_id   bigserial PRIMARY KEY,
  symbol     text NOT NULL,
  name       text,
  asset_type text NOT NULL,   -- ETF, CRYPTO, METAL, etc.
  unit       text,            -- oz, coin, share...
  base_ccy   text,            -- USD/EUR, etc.
  metadata   jsonb
);

CREATE UNIQUE INDEX IF NOT EXISTS ux_assets_symbol_type
ON n8n.assets(symbol, asset_type);

2.2 n8n.asset_ticks

Ticks are the hottest table in the system. The two goals are:

• insert safely even if a job retries (idempotency)
• fetch “latest tick” instantly

CREATE TABLE IF NOT EXISTS n8n.asset_ticks (
  ts_utc         timestamptz NOT NULL,
  asset_id       bigint NOT NULL REFERENCES n8n.assets(asset_id),
  provider       text NOT NULL,
  price_native   numeric(18,8),
  quote_ccy      text NOT NULL,     -- USD, EUR...
  price_eur      numeric(18,8),
  fx_used        numeric(12,6),
  fx_source      text,
  src_updated_at timestamptz,
  meta           jsonb
);

-- De-dupe: if the same asset/quote/provider/timestamp arrives again, treat it as the same tick
CREATE UNIQUE INDEX IF NOT EXISTS ux_ticks_dedup
ON n8n.asset_ticks(asset_id, quote_ccy, provider, ts_utc);

-- Make "latest USD tick per asset" fast (critical for bot + dashboards)
CREATE INDEX IF NOT EXISTS ix_ticks_usd_asset_ts_desc
ON n8n.asset_ticks(asset_id, ts_utc DESC)
WHERE quote_ccy = 'USD';

2.3 n8n.asset_daily

CREATE TABLE IF NOT EXISTS n8n.asset_daily (
  asset_id     bigint NOT NULL REFERENCES n8n.assets(asset_id),
  date_utc     date NOT NULL,
  close_eur    numeric(18,8),
  close_native numeric(18,8),
  PRIMARY KEY (asset_id, date_utc)
);

CREATE INDEX IF NOT EXISTS ix_daily_asset_date_desc
ON n8n.asset_daily(asset_id, date_utc DESC);

2.4 n8n.asset_news

News is where your LLM summary work lands. I strongly recommend URL de-dupe — it’s the simplest “don’t spam my DB with the same story” control.

CREATE TABLE IF NOT EXISTS n8n.asset_news (
  id            bigserial PRIMARY KEY,
  url           text NOT NULL,
  source_domain text,
  title         text,
  content_text  text,
  summary       text,
  quality_score numeric,
  confidence    numeric,
  published_at  timestamptz,
  fetched_at    timestamptz NOT NULL DEFAULT now(),
  asset_id      bigint NOT NULL REFERENCES n8n.assets(asset_id),
  asset_type    text,
  symbol        text,
  raw           jsonb
);

CREATE UNIQUE INDEX IF NOT EXISTS ux_news_url
ON n8n.asset_news(url);

CREATE INDEX IF NOT EXISTS ix_news_asset_published_desc
ON n8n.asset_news(asset_id, published_at DESC);

2.5 n8n.analyst_runs

This is the durable output of your “facts → signals” logic. Store the verdict + confidence, and keep the drivers in JSON so you can evolve features without schema churn.

CREATE TABLE IF NOT EXISTS n8n.analyst_runs (
  run_id         bigserial PRIMARY KEY,
  ts_utc         timestamptz NOT NULL DEFAULT now(),
  asset_id       bigint NOT NULL REFERENCES n8n.assets(asset_id),
  verdict        text,
  confidence_num numeric(5,2),
  drivers_json   jsonb,
  features_json  jsonb,
  news_json      jsonb,
  note           text
);

CREATE INDEX IF NOT EXISTS ix_analyst_asset_ts_desc
ON n8n.analyst_runs(asset_id, ts_utc DESC);

3) The Agent Query Contract (views for the Telegram copilot)

Dashboards are one consumer. A Telegram agent is another — and it’s much more sensitive to latency, surprises, and accidental complexity.

So I keep a deliberate set of agent-friendly views (n8n.v_*) that behave like a read-only contract:

• bounded windows (30 days, 24 hours)
• stable columns (consistent payload for message formatting)
• predictable performance (no huge scans)
• safe access control (grant SELECT only on views/MVs, deny raw tables)

These views are the “tools” the bot uses. They’re the guardrails.

3.1 n8n.v_asset_latest_usd — latest price per asset

CREATE OR REPLACE VIEW n8n.v_asset_latest_usd
AS
SELECT a.asset_id,
       a.symbol,
       t.ts_utc,
       t.price_native AS price_usd
FROM n8n.assets a
JOIN LATERAL (
  SELECT t_1.ts_utc,
         t_1.price_native
  FROM n8n.asset_ticks t_1
  WHERE t_1.asset_id = a.asset_id
    AND t_1.quote_ccy = 'USD'::text
  ORDER BY t_1.ts_utc DESC
  LIMIT 1
) t ON true;

Used for: “What’s the current BTC price?”, “Latest price of SXR8?” Why it works: one row per asset, fast when ix_ticks_usd_asset_ts_desc exists.

3.2 n8n.v_asset_daily30_usd — recent daily closes (USD)

CREATE OR REPLACE VIEW n8n.v_asset_daily30_usd
AS
SELECT ad.asset_id,
       a.symbol,
       ad.date_utc AS d,
       ad.close_native
FROM n8n.asset_daily ad
JOIN n8n.assets a ON a.asset_id = ad.asset_id
WHERE ad.date_utc >= (CURRENT_DATE - '35 days'::interval)
  AND ad.close_native IS NOT NULL;

Used for: “Show me BTC trend for the last month” Why it matters: forces a bounded window so the agent can’t accidentally query all history.

3.3 n8n.v_asset_daily30 — unified close value for mixed assets

CREATE OR REPLACE VIEW n8n.v_asset_daily30
AS
SELECT ad.asset_id,
       a.symbol,
       ad.date_utc AS d,
       COALESCE(ad.close_native, ad.close_eur) AS close_val
FROM n8n.asset_daily ad
JOIN n8n.assets a ON a.asset_id = ad.asset_id
WHERE ad.date_utc >= (CURRENT_DATE - '35 days'::interval);

Used for: “Compare last month's performance” Why it matters: the bot doesn’t need to care whether the asset uses native close or EUR close.

3.4 n8n.v_news_24h — strict 24h news window

CREATE OR REPLACE VIEW n8n.v_news_24h
AS
SELECT asset_news.id,
       asset_news.url,
       asset_news.source_domain,
       asset_news.title,
       asset_news.content_text,
       asset_news.summary,
       asset_news.quality_score,
       asset_news.confidence,
       asset_news.published_at,
       asset_news.fetched_at,
       asset_news.asset_id,
       asset_news.asset_type,
       asset_news.symbol,
       asset_news.raw
FROM n8n.asset_news
WHERE asset_news.published_at >= (now() - '24:00:00'::interval);

Used for: “What happened in the last 24h for BTC?” Why it matters: keeps answers relevant and query cost bounded.

3.5 n8n.v_news_recent — clean payload for Telegram messages

CREATE OR REPLACE VIEW n8n.v_news_recent
AS
SELECT n.asset_id,
       a.symbol,
       n.title,
       n.source_domain,
       n.url,
       n.published_at,
       COALESCE(n.summary, ''::text) AS summary
FROM n8n.asset_news n
JOIN n8n.assets a ON a.asset_id = n.asset_id;

Used for: headline lists and “give me the latest 5 stories” outputs. Why it matters: consistent formatting fields for Telegram cards.

Big-system trick: later, you can transparently re-point these `v_views tomv_` tables for speed without changing the agent at all. The view becomes the stable interface; the implementation can evolve.

4) Materialised views for near-real-time reporting

Views make the agent safe and predictable. Materialised views make dashboards fast and stable.

I treat MVs as “dashboard primitives”:

• latest snapshot tables
• time-bucketed OHLC for charts
• rolling returns
• news rollups
• latest analyst outputs
• feed health / freshness

4.1 n8n.mv_asset_latest_usd — KPI-friendly latest price snapshot

CREATE MATERIALIZED VIEW IF NOT EXISTS n8n.mv_asset_latest_usd AS
SELECT
  a.asset_id,
  a.symbol,
  t.ts_utc,
  t.price_native AS price_usd,
  now() - t.ts_utc AS age
FROM n8n.assets a
JOIN LATERAL (
  SELECT t1.ts_utc, t1.price_native
  FROM n8n.asset_ticks t1
  WHERE t1.asset_id = a.asset_id
    AND t1.quote_ccy = 'USD'
  ORDER BY t1.ts_utc DESC
  LIMIT 1
) t ON true;

CREATE UNIQUE INDEX IF NOT EXISTS ux_mv_asset_latest_usd
ON n8n.mv_asset_latest_usd(asset_id);

4.2 n8n.mv_asset_ohlc_5m_usd_48h — intraday candles (last 48h)

CREATE MATERIALIZED VIEW IF NOT EXISTS n8n.mv_asset_ohlc_5m_usd_48h AS
WITH base AS (
  SELECT
    asset_id,
    date_bin(INTERVAL '5 minutes', ts_utc, TIMESTAMPTZ '1970-01-01') AS bucket_ts,
    ts_utc,
    price_native
  FROM n8n.asset_ticks
  WHERE quote_ccy = 'USD'
    AND ts_utc >= now() - INTERVAL '48 hours'
),
agg AS (
  SELECT
    asset_id,
    bucket_ts,
    MIN(price_native) AS low,
    MAX(price_native) AS high,
    (ARRAY_AGG(price_native ORDER BY ts_utc ASC))[1]  AS open,
    (ARRAY_AGG(price_native ORDER BY ts_utc DESC))[1] AS close
  FROM base
  GROUP BY asset_id, bucket_ts
)
SELECT
  a.asset_id,
  s.symbol,
  a.bucket_ts,
  a.open, a.high, a.low, a.close
FROM agg a
JOIN n8n.assets s ON s.asset_id = a.asset_id;

CREATE UNIQUE INDEX IF NOT EXISTS ux_mv_asset_ohlc_5m_usd_48h
ON n8n.mv_asset_ohlc_5m_usd_48h(asset_id, bucket_ts);

4.3 n8n.mv_asset_returns_365d — latest 1D/7D/30D returns

CREATE MATERIALIZED VIEW IF NOT EXISTS n8n.mv_asset_returns_365d AS
WITH d AS (
  SELECT
    ad.asset_id,
    a.symbol,
    ad.date_utc,
    ad.close_native AS close_usd
  FROM n8n.asset_daily ad
  JOIN n8n.assets a ON a.asset_id = ad.asset_id
  WHERE ad.close_native IS NOT NULL
    AND ad.date_utc >= CURRENT_DATE - 365
),
x AS (
  SELECT
    asset_id,
    symbol,
    date_utc,
    close_usd,
    LAG(close_usd, 1)  OVER (PARTITION BY asset_id ORDER BY date_utc) AS close_1d_ago,
    LAG(close_usd, 7)  OVER (PARTITION BY asset_id ORDER BY date_utc) AS close_7d_ago,
    LAG(close_usd, 30) OVER (PARTITION BY asset_id ORDER BY date_utc) AS close_30d_ago
  FROM d
)
SELECT DISTINCT ON (asset_id)
  asset_id,
  symbol,
  date_utc AS as_of_date,
  close_usd,
  (close_usd / close_1d_ago  - 1) AS ret_1d,
  (close_usd / close_7d_ago  - 1) AS ret_7d,
  (close_usd / close_30d_ago - 1) AS ret_30d
FROM x
ORDER BY asset_id, date_utc DESC;

CREATE UNIQUE INDEX IF NOT EXISTS ux_mv_asset_returns_365d
ON n8n.mv_asset_returns_365d(asset_id);

4.4 News rollups + latest article per asset

CREATE MATERIALIZED VIEW IF NOT EXISTS n8n.mv_news_volume_24h_hourly AS
SELECT
  n.asset_id,
  a.symbol,
  date_trunc('hour', n.published_at) AS hour_ts,
  COUNT(*) AS articles,
  AVG(n.quality_score) AS avg_quality,
  AVG(n.confidence)    AS avg_confidence
FROM n8n.asset_news n
JOIN n8n.assets a ON a.asset_id = n.asset_id
WHERE n.published_at >= now() - INTERVAL '24 hours'
GROUP BY n.asset_id, a.symbol, date_trunc('hour', n.published_at);

CREATE UNIQUE INDEX IF NOT EXISTS ux_mv_news_volume_24h_hourly
ON n8n.mv_news_volume_24h_hourly(asset_id, hour_ts);

CREATE MATERIALIZED VIEW IF NOT EXISTS n8n.mv_news_latest_per_asset AS
SELECT DISTINCT ON (n.asset_id)
  n.asset_id,
  a.symbol,
  n.published_at,
  n.title,
  n.source_domain,
  n.url,
  COALESCE(n.summary, '') AS summary,
  n.quality_score,
  n.confidence
FROM n8n.asset_news n
JOIN n8n.assets a ON a.asset_id = n.asset_id
ORDER BY n.asset_id, n.published_at DESC;

CREATE UNIQUE INDEX IF NOT EXISTS ux_mv_news_latest_per_asset
ON n8n.mv_news_latest_per_asset(asset_id);

4.5 Latest analyst verdict per asset

CREATE MATERIALIZED VIEW IF NOT EXISTS n8n.mv_analyst_latest AS
SELECT DISTINCT ON (ar.asset_id)
  ar.asset_id,
  a.symbol,
  ar.ts_utc,
  ar.verdict,
  ar.confidence_num,
  ar.drivers_json,
  ar.note
FROM n8n.analyst_runs ar
JOIN n8n.assets a ON a.asset_id = ar.asset_id
ORDER BY ar.asset_id, ar.ts_utc DESC;

CREATE UNIQUE INDEX IF NOT EXISTS ux_mv_analyst_latest
ON n8n.mv_analyst_latest(asset_id);

4.6 n8n.mv_feed_health — “Is the system alive?” (fast, no join explosion)

This MV is a common trap if you implement it as a join across multiple large tables. The correct pattern is “latest row per asset” using indexed lookups.

CREATE MATERIALIZED VIEW IF NOT EXISTS n8n.mv_feed_health AS
SELECT
  a.asset_id,
  a.symbol,

  t.last_tick_ts,
  d.last_daily_date,
  n.last_news_ts,
  ar.last_analyst_ts,

  now() - t.last_tick_ts AS tick_lag,
  now() - n.last_news_ts AS news_lag
FROM n8n.assets a

LEFT JOIN LATERAL (
  SELECT t1.ts_utc AS last_tick_ts
  FROM n8n.asset_ticks t1
  WHERE t1.asset_id = a.asset_id
    AND t1.quote_ccy = 'USD'
  ORDER BY t1.ts_utc DESC
  LIMIT 1
) t ON true

LEFT JOIN LATERAL (
  SELECT d1.date_utc AS last_daily_date
  FROM n8n.asset_daily d1
  WHERE d1.asset_id = a.asset_id
  ORDER BY d1.date_utc DESC
  LIMIT 1
) d ON true

LEFT JOIN LATERAL (
  SELECT n1.published_at AS last_news_ts
  FROM n8n.asset_news n1
  WHERE n1.asset_id = a.asset_id
  ORDER BY n1.published_at DESC
  LIMIT 1
) n ON true

LEFT JOIN LATERAL (
  SELECT ar1.ts_utc AS last_analyst_ts
  FROM n8n.analyst_runs ar1
  WHERE ar1.asset_id = a.asset_id
  ORDER BY ar1.ts_utc DESC
  LIMIT 1
) ar ON true;

CREATE UNIQUE INDEX IF NOT EXISTS ux_mv_feed_health
ON n8n.mv_feed_health(asset_id);

5) Refresh strategy (cadence + automation)

Refresh cadence (what I actually do)

• mv_asset_latest_usd → every 1–2 minutes
• mv_feed_health → every 1 minute
• mv_asset_ohlc_5m_usd_48h → every 5 minutes
• mv_news_latest_per_asset → every 5 minutes
• mv_news_volume_24h_hourly → every 5–10 minutes
• mv_analyst_latest → every 2–5 minutes (or right after an analyst run)
• mv_asset_returns_365d → daily (after daily close)

Refresh all materialised views in n8n (with locking + concurrent refresh)

CREATE OR REPLACE PROCEDURE n8n.refresh_all_mviews(p_concurrently boolean DEFAULT true)
LANGUAGE plpgsql
AS $$
DECLARE
  r record;
  sql text;
BEGIN
  IF NOT pg_try_advisory_lock(hashtext('n8n.refresh_all_mviews')) THEN
    RAISE NOTICE 'Another refresh job is already running. Skipping.';
    RETURN;
  END IF;

  FOR r IN
    SELECT schemaname, matviewname
    FROM pg_matviews
    WHERE schemaname = 'n8n'
    ORDER BY matviewname
  LOOP
    BEGIN
      IF p_concurrently THEN
        sql := format('REFRESH MATERIALIZED VIEW CONCURRENTLY %I.%I', r.schemaname, r.matviewname);
      ELSE
        sql := format('REFRESH MATERIALIZED VIEW %I.%I', r.schemaname, r.matviewname);
      END IF;

      RAISE NOTICE 'Refreshing: %', sql;
      EXECUTE sql;

    EXCEPTION WHEN OTHERS THEN
      RAISE WARNING 'Failed refreshing %.% : %', r.schemaname, r.matviewname, SQLERRM;
    END;
  END LOOP;

  PERFORM pg_advisory_unlock(hashtext('n8n.refresh_all_mviews'));
END;
$$;

-- Run:
-- CALL n8n.refresh_all_mviews(true);

Why CONCURRENTLY matters: dashboards can keep reading while the refresh runs (as long as each MV has a unique index).

7) Retention (keeping ticks under control)

Ticks grow fast. Retention is not optional if you’re collecting frequently.

A simple policy to start:

• Keep raw ticks for 60 days
• keep daily closes indefinitely
• Keep news and analyst outputs as long as you care about backtesting/explanations

Example cleanup job:

DELETE FROM n8n.asset_ticks
WHERE ts_utc < now() - INTERVAL '60 days';

Later, if ticks become huge, the next upgrade is partitioning — but that’s a “future improvements” topic.

8) Wrap-up: from “tables” to a real-time query surface

This is the point where Postgres stops being a dumping ground for feeds and starts behaving like a proper query layer for the whole copilot.

The raw tables (assets, asset_ticks, asset_daily, asset_news, analyst_runs) stay deliberately simple so n8n can insert continuously without drama. On top of that, the v_* views act as a stable contract for the Telegram agent — bounded time windows, predictable columns, and a safe place for “latest price / last 30 days / last 24 hours” questions to land without the bot ever touching raw ingestion tables directly. Finally, the mv_* materialised views turn the expensive stuff into precomputed primitives so dashboards and chat queries stay snappy even when the pipelines are hot.

In Part 8, I’m going to build the “live cockpit” on top of this: near-real-time charts that load instantly, a clean market overview (latest + movers), news panels that connect price moves to headlines, and a freshness/exception layer that tells you when a feed is stale before you notice it manually. The goal is to make the system feel less like a set of scripts and more like an operator-grade product you can trust day to day.

Read the next article