Introduce PriceSource and PriceSink architecture for AssetAZ Ticker #61

Closed
opened 2026-08-06 21:53:24 +02:00 by minimons · 0 comments
Owner

Background

Issue #60 introduced the first AssetAZ Ticker implementation with one internally hardcoded EVE_USDC price source.

The hardcoded implementation proved the complete flow:

  1. Produce a price observation.
  2. Persist it.
  3. Publish it through TickerService only after successful persistence.

The next step is to separate price acquisition from the Ticker itself.

The Ticker must not know whether a price is obtained through polling, a WebSocket stream, manual entry or another mechanism. Concrete price sources shall be plugins which announce observations through a small sink interface.

This issue introduces that general architecture while retaining the existing hardcoded EVE_USDC = 14.85 source. A real Raydium source will be implemented in a separate issue.

Goal

Refactor the AssetAZ Ticker so that:

  • price acquisition is performed by independent PriceSource implementations;
  • sources announce prices through a PriceSink;
  • TickerServiceImpl implements both TickerService and PriceSink;
  • the existing hardcoded price becomes the first PriceSource implementation;
  • every combination of TradingPair and source has its own history file;
  • getLatestPrice(tradingPair) returns the newest successfully persisted observation across all active sources for that pair;
  • the existing persist-before-publish guarantee is preserved.

After this issue, the externally visible behavior shall remain the same: the Ticker produces EVE_USDC = 14.85 immediately and then once per minute.

API

Add the following interfaces under:

com.r35157.assetaz.core.service.ticker

PriceSource

The interface shall be equivalent to:

public interface PriceSource {

    @NotNull
    TradingPair getTradingPair();

    @NotNull
    ΩPriceSourceNameΩ getSourceName();

    void start();

    void stop();
}

A source owns the mechanism used to obtain prices, including any scheduler, polling thread, WebSocket client or other required resource.

The source receives a PriceSink during construction and announces observations through that reference.

getTradingPair() and getSourceName() together identify one concrete source history.

The source name must be stable across restarts. For example, a future Raydium source may use:

Raydium_EpZXeeShz64DK8HfKnjWPTkmhN7dZe77SjvzNDX8CARq

PriceSink

Add a functional interface equivalent to:

@FunctionalInterface
public interface PriceSink {

    void announce(
            @NotNull PriceSource source,
            @NotNull ΩPriceΩ price,
            @NotNull Instant observedAt
    );
}

A source shall normally announce itself:

priceSink.announce(this, price, observedAt);

The source supplies the raw typed price. The Ticker constructs the corresponding AssetPrice using the source's registered TradingPair.

This avoids duplicating the trading pair in both the source metadata and callback parameters.

TickerService

Extend the Ticker lifecycle and source registration API as needed so that sources can be registered before startup and stopped with the Ticker.

An API equivalent to the following is acceptable:

public interface TickerService {

    void addPriceSource(@NotNull PriceSource priceSource);

    void start();

    void stop();

    @NotNull
    PriceObservation getLatestPrice(@NotNull TradingPair tradingPair);
}

TickerService shall not extend PriceSink. Price-producing plugins shall receive only a PriceSink reference.

A source must be registered before the Ticker is started. Duplicate registrations with the same TradingPair and source name shall be rejected clearly because they would address the same history.

Price source name ValueTag

Add the following ValueTag under the existing String -> Name hierarchy in the Detag configuration:

String
    Name
        PriceSourceName

Its generated Java backing type is String.

Use it in the PriceSource contract:

@NotNull ΩPriceSourceNameΩ getSourceName();

A source name is part of the persistent identity and must be valid as one directory component.

Reject source names which are empty or contain path traversal or unsafe directory content, including:

  • /
  • \
  • ..
  • control characters

Do not silently rewrite a source name, because changing it would address a different history.

PriceObservation

Extend the in-memory observation type with its source identity:

public record PriceObservation(
        @NotNull AssetPrice price,
        @NotNull Instant observedAt,
        @NotNull ΩPriceSourceNameΩ sourceName
) {
}

The source name belongs in the in-memory observation so callers can see which source supplied the returned price.

It shall not be added to individual lines in the persisted history file. The directory containing the history file identifies the source.

Reference implementation

Change:

com.r35157.assetaz.core.service.ticker.impl.ref.TickerServiceImpl

so it implements:

TickerService, PriceSink

Move all price-generation responsibilities out of TickerServiceImpl.

Create a hardcoded PriceSource implementation in the reference implementation package. A suitable name is:

HardcodedPriceSource

It shall:

  • use EVE_USDC;
  • use the source name Hardcoded;
  • announce price 14.85;
  • announce once immediately when started;
  • wait one minute after an attempt completes before making the next attempt;
  • use fixed-delay behavior rather than fixed-rate scheduling;
  • own and stop its scheduler;
  • receive only a PriceSink reference for publishing observations.

The scheduler must therefore move from TickerServiceImpl into HardcodedPriceSource.

TickerServiceImpl must not contain knowledge of the hardcoded price, polling interval or scheduling mechanism.

Source lifecycle

When the Ticker starts, it shall:

  1. Validate all registered source identities.
  2. Determine the history path for each registered source.
  3. Activate only sources whose expected history file already exists.
  4. Load all active source histories.
  5. determine the latest persisted observation for each trading pair across all active sources;
  6. start each active PriceSource.

When the Ticker stops, it shall stop every source that it started.

A source whose history file is missing shall not be started. The Ticker shall log a warning containing:

  • the trading pair;
  • the source name;
  • the complete expected history path.

A missing history file or directory must not be created automatically.

Only registered sources shall be considered. The Ticker shall not discover and activate arbitrary source directories by scanning the filesystem.

Persistent directory structure

Replace the old single-file layout:

data/assetaz/ticker/EVE_USDC.prices

with:

data/assetaz/ticker/
└── <base UUID>/
    └── <quote UUID>/
        └── <source name>/
            └── <base symbol>_<quote symbol>.prices

The UUIDs are the technical identity. Their canonical representation, including hyphens, must remain unchanged.

The symbols are included only to make the final filename understandable to a human.

For the hardcoded EVE_USDC source, the path is:

data/assetaz/ticker/
└── 019c3f9f-41d1-7a73-b1df-d4c11c7ff301/
    └── 019c3f9f-41d1-7a73-b1df-d4c11c7ff302/
        └── Hardcoded/
            └── EVE_USDC.prices

Path construction shall use:

  • source.getTradingPair().base().id() for the first UUID;
  • source.getTradingPair().quote().id() for the second UUID;
  • source.getSourceName() for the source directory;
  • the base and quote symbols for the human-readable filename.

Centralize this path construction in the Ticker implementation so concrete sources never construct or know their persistence paths.

Symbols used in the final filename shall be converted to safe filename components. Symbol text is not used as technical identity and must not replace the UUIDs.

History format

Keep the existing line format unchanged:

<UTC timestamp>:<price>

with timestamp format:

uuuuMMddHHmmssSSS'Z'

Example:

20260805131542783Z:14.85

Do not store the source name or trading pair in each line.

Continue supporting:

  • empty lines;
  • full-line comments;
  • inline comments;
  • strict error reporting for malformed lines.

Automatic writes shall append plain observation lines and preserve existing comments and blank lines.

When loading a history, construct each PriceObservation using the TradingPair and source name belonging to the registered source whose history is being loaded.

Source activation

Activation now applies to a specific combination of TradingPair and PriceSource.

An existing history file activates that source:

  • A non-empty file restores all valid observations.
  • An empty or comment-only file activates the source without an initial price.
  • A missing file leaves only that source inactive.
  • Other valid sources for the same trading pair may remain active.

The Ticker shall not create missing directories or history files automatically.

Announcing and persisting observations

When an active, registered source calls PriceSink.announce(...), the Ticker shall:

  1. Validate that the callback came from a registered and active source.
  2. Validate the price and timestamp.
  3. Construct an AssetPrice using the announced price and the source's registered TradingPair.
  4. Construct a PriceObservation containing the AssetPrice, timestamp and source name.
  5. Append the observation to that source's history file.
  6. Flush or force the data using the existing durable FileChannel.force(true) behavior.
  7. Only after persistence succeeds, consider the observation for the in-memory latest value.

If persistence fails:

  • the observation must not become visible through getLatestPrice(...);
  • the previously persisted latest observation must remain available;
  • the failure must be reported clearly with the trading pair, source and history path.

Callbacks from different sources may occur concurrently. The implementation must prevent corrupted or interleaved writes and ensure that readers never observe an unpersisted observation.

Latest price across sources

Keep the existing API:

@NotNull
PriceObservation getLatestPrice(@NotNull TradingPair tradingPair);

If several active sources provide the same trading pair, return the successfully persisted observation with the greatest observedAt timestamp across all those sources.

The source name in the returned PriceObservation identifies which source supplied it.

A persisted observation shall still be appended even when an already loaded observation has a later timestamp. In that case, the later existing observation remains the latest.

Continue throwing clear exceptions for:

  • a trading pair with no registered sources;
  • a trading pair with no active sources;
  • an active trading pair for which no observation has yet been persisted.

Startup wiring

Update the temporary hardcoded startup in NenjimHubImpl so that it:

  1. Constructs one TickerServiceImpl.
  2. Constructs one HardcodedPriceSource with the Ticker implementation passed as PriceSink.
  3. Registers the source with the Ticker.
  4. Starts the Ticker.

This remains temporary Cauldron wiring and does not establish ownership of AssetAZ by NenjimHub.

Migration

The old history file:

data/assetaz/ticker/EVE_USDC.prices

shall no longer be read after this change.

It must be moved by the operator to:

data/assetaz/ticker/019c3f9f-41d1-7a73-b1df-d4c11c7ff301/019c3f9f-41d1-7a73-b1df-d4c11c7ff302/Hardcoded/EVE_USDC.prices

Do not automatically migrate, copy or create the history. Automatic migration could hide an incorrect working directory or accidentally split one history into multiple files.

Acceptance criteria

  • PriceSource and functional PriceSink interfaces exist in the AssetAZ Ticker API package.
  • ΩPriceSourceNameΩ exists under String -> Name.
  • TickerServiceImpl implements both TickerService and PriceSink.
  • The hardcoded generator is a separate PriceSource.
  • TickerServiceImpl contains no hardcoded price, one-minute interval or price-generation scheduler.
  • The hardcoded source still announces EVE_USDC = 14.85 immediately and then with one-minute fixed delay.
  • The hardcoded source can be stopped without leaving its scheduler running.
  • Each source has an independent history under the UUID-based directory structure.
  • The source name is stored in PriceObservation, but not in history lines.
  • Missing source histories are not created and their sources are not started.
  • Existing empty source histories activate their sources.
  • Existing histories load source identity from the registered source and directory context.
  • Duplicate (TradingPair, sourceName) registrations are rejected.
  • Persistence occurs before publication.
  • Persistence failure preserves the previous latest observation.
  • getLatestPrice(tradingPair) selects the greatest persisted timestamp across all active sources for that pair.
  • The project compiles and existing behavior unrelated to the Ticker remains unchanged.

Out of scope

  • Raydium integration.
  • Any other real market-price source.
  • Changes to Raydium.fetchPoolPrice(...).
  • WebSocket or streaming implementation.
  • Additional trading pairs beyond the existing EVE_USDC hardcoded source.
  • Source-specific price-query methods.
  • Public history-query APIs.
  • Retention, compaction or downsampling.
  • Integration with Evelyn, alarms, EMC or portfolio services.
## Background Issue #60 introduced the first AssetAZ Ticker implementation with one internally hardcoded `EVE_USDC` price source. The hardcoded implementation proved the complete flow: 1. Produce a price observation. 2. Persist it. 3. Publish it through `TickerService` only after successful persistence. The next step is to separate price acquisition from the Ticker itself. The Ticker must not know whether a price is obtained through polling, a WebSocket stream, manual entry or another mechanism. Concrete price sources shall be plugins which announce observations through a small sink interface. This issue introduces that general architecture while retaining the existing hardcoded `EVE_USDC = 14.85` source. A real Raydium source will be implemented in a separate issue. ## Goal Refactor the AssetAZ Ticker so that: * price acquisition is performed by independent `PriceSource` implementations; * sources announce prices through a `PriceSink`; * `TickerServiceImpl` implements both `TickerService` and `PriceSink`; * the existing hardcoded price becomes the first `PriceSource` implementation; * every combination of `TradingPair` and source has its own history file; * `getLatestPrice(tradingPair)` returns the newest successfully persisted observation across all active sources for that pair; * the existing persist-before-publish guarantee is preserved. After this issue, the externally visible behavior shall remain the same: the Ticker produces `EVE_USDC = 14.85` immediately and then once per minute. ## API Add the following interfaces under: ```java com.r35157.assetaz.core.service.ticker ``` ### `PriceSource` The interface shall be equivalent to: ```java public interface PriceSource { @NotNull TradingPair getTradingPair(); @NotNull ΩPriceSourceNameΩ getSourceName(); void start(); void stop(); } ``` A source owns the mechanism used to obtain prices, including any scheduler, polling thread, WebSocket client or other required resource. The source receives a `PriceSink` during construction and announces observations through that reference. `getTradingPair()` and `getSourceName()` together identify one concrete source history. The source name must be stable across restarts. For example, a future Raydium source may use: ```text Raydium_EpZXeeShz64DK8HfKnjWPTkmhN7dZe77SjvzNDX8CARq ``` ### `PriceSink` Add a functional interface equivalent to: ```java @FunctionalInterface public interface PriceSink { void announce( @NotNull PriceSource source, @NotNull ΩPriceΩ price, @NotNull Instant observedAt ); } ``` A source shall normally announce itself: ```java priceSink.announce(this, price, observedAt); ``` The source supplies the raw typed price. The Ticker constructs the corresponding `AssetPrice` using the source's registered `TradingPair`. This avoids duplicating the trading pair in both the source metadata and callback parameters. ### `TickerService` Extend the Ticker lifecycle and source registration API as needed so that sources can be registered before startup and stopped with the Ticker. An API equivalent to the following is acceptable: ```java public interface TickerService { void addPriceSource(@NotNull PriceSource priceSource); void start(); void stop(); @NotNull PriceObservation getLatestPrice(@NotNull TradingPair tradingPair); } ``` `TickerService` shall not extend `PriceSink`. Price-producing plugins shall receive only a `PriceSink` reference. A source must be registered before the Ticker is started. Duplicate registrations with the same `TradingPair` and source name shall be rejected clearly because they would address the same history. ## Price source name ValueTag Add the following ValueTag under the existing `String -> Name` hierarchy in the Detag configuration: ```text String Name PriceSourceName ``` Its generated Java backing type is `String`. Use it in the `PriceSource` contract: ```java @NotNull ΩPriceSourceNameΩ getSourceName(); ``` A source name is part of the persistent identity and must be valid as one directory component. Reject source names which are empty or contain path traversal or unsafe directory content, including: * `/` * `\` * `..` * control characters Do not silently rewrite a source name, because changing it would address a different history. ## `PriceObservation` Extend the in-memory observation type with its source identity: ```java public record PriceObservation( @NotNull AssetPrice price, @NotNull Instant observedAt, @NotNull ΩPriceSourceNameΩ sourceName ) { } ``` The source name belongs in the in-memory observation so callers can see which source supplied the returned price. It shall not be added to individual lines in the persisted history file. The directory containing the history file identifies the source. ## Reference implementation Change: ```java com.r35157.assetaz.core.service.ticker.impl.ref.TickerServiceImpl ``` so it implements: ```java TickerService, PriceSink ``` Move all price-generation responsibilities out of `TickerServiceImpl`. Create a hardcoded `PriceSource` implementation in the reference implementation package. A suitable name is: ```java HardcodedPriceSource ``` It shall: * use `EVE_USDC`; * use the source name `Hardcoded`; * announce price `14.85`; * announce once immediately when started; * wait one minute after an attempt completes before making the next attempt; * use fixed-delay behavior rather than fixed-rate scheduling; * own and stop its scheduler; * receive only a `PriceSink` reference for publishing observations. The scheduler must therefore move from `TickerServiceImpl` into `HardcodedPriceSource`. `TickerServiceImpl` must not contain knowledge of the hardcoded price, polling interval or scheduling mechanism. ## Source lifecycle When the Ticker starts, it shall: 1. Validate all registered source identities. 2. Determine the history path for each registered source. 3. Activate only sources whose expected history file already exists. 4. Load all active source histories. 5. determine the latest persisted observation for each trading pair across all active sources; 6. start each active `PriceSource`. When the Ticker stops, it shall stop every source that it started. A source whose history file is missing shall not be started. The Ticker shall log a warning containing: * the trading pair; * the source name; * the complete expected history path. A missing history file or directory must not be created automatically. Only registered sources shall be considered. The Ticker shall not discover and activate arbitrary source directories by scanning the filesystem. ## Persistent directory structure Replace the old single-file layout: ```text data/assetaz/ticker/EVE_USDC.prices ``` with: ```text data/assetaz/ticker/ └── <base UUID>/ └── <quote UUID>/ └── <source name>/ └── <base symbol>_<quote symbol>.prices ``` The UUIDs are the technical identity. Their canonical representation, including hyphens, must remain unchanged. The symbols are included only to make the final filename understandable to a human. For the hardcoded `EVE_USDC` source, the path is: ```text data/assetaz/ticker/ └── 019c3f9f-41d1-7a73-b1df-d4c11c7ff301/ └── 019c3f9f-41d1-7a73-b1df-d4c11c7ff302/ └── Hardcoded/ └── EVE_USDC.prices ``` Path construction shall use: * `source.getTradingPair().base().id()` for the first UUID; * `source.getTradingPair().quote().id()` for the second UUID; * `source.getSourceName()` for the source directory; * the base and quote symbols for the human-readable filename. Centralize this path construction in the Ticker implementation so concrete sources never construct or know their persistence paths. Symbols used in the final filename shall be converted to safe filename components. Symbol text is not used as technical identity and must not replace the UUIDs. ## History format Keep the existing line format unchanged: ```text <UTC timestamp>:<price> ``` with timestamp format: ```text uuuuMMddHHmmssSSS'Z' ``` Example: ```text 20260805131542783Z:14.85 ``` Do not store the source name or trading pair in each line. Continue supporting: * empty lines; * full-line comments; * inline comments; * strict error reporting for malformed lines. Automatic writes shall append plain observation lines and preserve existing comments and blank lines. When loading a history, construct each `PriceObservation` using the `TradingPair` and source name belonging to the registered source whose history is being loaded. ## Source activation Activation now applies to a specific combination of `TradingPair` and `PriceSource`. An existing history file activates that source: * A non-empty file restores all valid observations. * An empty or comment-only file activates the source without an initial price. * A missing file leaves only that source inactive. * Other valid sources for the same trading pair may remain active. The Ticker shall not create missing directories or history files automatically. ## Announcing and persisting observations When an active, registered source calls `PriceSink.announce(...)`, the Ticker shall: 1. Validate that the callback came from a registered and active source. 2. Validate the price and timestamp. 3. Construct an `AssetPrice` using the announced price and the source's registered `TradingPair`. 4. Construct a `PriceObservation` containing the `AssetPrice`, timestamp and source name. 5. Append the observation to that source's history file. 6. Flush or force the data using the existing durable `FileChannel.force(true)` behavior. 7. Only after persistence succeeds, consider the observation for the in-memory latest value. If persistence fails: * the observation must not become visible through `getLatestPrice(...)`; * the previously persisted latest observation must remain available; * the failure must be reported clearly with the trading pair, source and history path. Callbacks from different sources may occur concurrently. The implementation must prevent corrupted or interleaved writes and ensure that readers never observe an unpersisted observation. ## Latest price across sources Keep the existing API: ```java @NotNull PriceObservation getLatestPrice(@NotNull TradingPair tradingPair); ``` If several active sources provide the same trading pair, return the successfully persisted observation with the greatest `observedAt` timestamp across all those sources. The source name in the returned `PriceObservation` identifies which source supplied it. A persisted observation shall still be appended even when an already loaded observation has a later timestamp. In that case, the later existing observation remains the latest. Continue throwing clear exceptions for: * a trading pair with no registered sources; * a trading pair with no active sources; * an active trading pair for which no observation has yet been persisted. ## Startup wiring Update the temporary hardcoded startup in `NenjimHubImpl` so that it: 1. Constructs one `TickerServiceImpl`. 2. Constructs one `HardcodedPriceSource` with the Ticker implementation passed as `PriceSink`. 3. Registers the source with the Ticker. 4. Starts the Ticker. This remains temporary Cauldron wiring and does not establish ownership of AssetAZ by NenjimHub. ## Migration The old history file: ```text data/assetaz/ticker/EVE_USDC.prices ``` shall no longer be read after this change. It must be moved by the operator to: ```text data/assetaz/ticker/019c3f9f-41d1-7a73-b1df-d4c11c7ff301/019c3f9f-41d1-7a73-b1df-d4c11c7ff302/Hardcoded/EVE_USDC.prices ``` Do not automatically migrate, copy or create the history. Automatic migration could hide an incorrect working directory or accidentally split one history into multiple files. ## Acceptance criteria * `PriceSource` and functional `PriceSink` interfaces exist in the AssetAZ Ticker API package. * `ΩPriceSourceNameΩ` exists under `String -> Name`. * `TickerServiceImpl` implements both `TickerService` and `PriceSink`. * The hardcoded generator is a separate `PriceSource`. * `TickerServiceImpl` contains no hardcoded price, one-minute interval or price-generation scheduler. * The hardcoded source still announces `EVE_USDC = 14.85` immediately and then with one-minute fixed delay. * The hardcoded source can be stopped without leaving its scheduler running. * Each source has an independent history under the UUID-based directory structure. * The source name is stored in `PriceObservation`, but not in history lines. * Missing source histories are not created and their sources are not started. * Existing empty source histories activate their sources. * Existing histories load source identity from the registered source and directory context. * Duplicate `(TradingPair, sourceName)` registrations are rejected. * Persistence occurs before publication. * Persistence failure preserves the previous latest observation. * `getLatestPrice(tradingPair)` selects the greatest persisted timestamp across all active sources for that pair. * The project compiles and existing behavior unrelated to the Ticker remains unchanged. ## Out of scope * Raydium integration. * Any other real market-price source. * Changes to `Raydium.fetchPoolPrice(...)`. * WebSocket or streaming implementation. * Additional trading pairs beyond the existing `EVE_USDC` hardcoded source. * Source-specific price-query methods. * Public history-query APIs. * Retention, compaction or downsampling. * Integration with Evelyn, alarms, EMC or portfolio services.
minimons added the enhancement label 2026-08-06 21:53:24 +02:00
minimons self-assigned this 2026-08-06 21:53:24 +02:00
minimons added this to the AssetAZ project 2026-08-06 21:53:24 +02:00
Sign in to join this conversation.
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: r35157/com_r35157_nenjim-hubd-impl_ref#61